From b34509922a272f464a2b0e09a73c6a2d8a6f0088 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 25 Oct 2026 12:09:02 -0500 Subject: [PATCH] [ruby] open no savepoints inside a transaction The Ruby scheduler opens a savepoint for each due job, up to 1,000 per pass. It uses the savepoint to catch a unique key conflict, roll back that one row, and mark the row discarded. In Postgres, a savepoint gets its own transaction ID when it first writes or locks a row. Each backend caches at most 64 of these IDs. A released savepoint keeps its entry until the top-level transaction ends, so a pass that makes more than 64 jobs available overflows the cache. Until that transaction ends, snapshots that any session takes are marked as overflowed. Their visibility checks on rows that recent transactions wrote must then look up `pg_subtrans`. This can slow queries across the server. The driver `transaction` helpers have the same problem. The Sequel helper passes `savepoint: true` and the Active Record helper passes `requires_new: true`, so inside an application transaction each River write can open a savepoint. An insert opens two. Replace the scheduler's per-job savepoints with a port of Go's `JobSchedule`. On Postgres, one statement locks the due rows with `FOR UPDATE`, finds conflicts with `jobs_with_rownum` and `unique_conflicts`, and sets each row to `available` or `discarded`. The query is equivalent to Go's and locks rows the same way, so the scheduler now waits for a row that another transaction has locked instead of skipping it. On SQLite, the pass follows Go's SQLite driver and checks each unique job for a conflict before it writes. Neither path relies on a failed write to find a conflict, so neither needs a savepoint. The pass now also notifies a queue whose only changed job is discarded, as Go does. Make both helpers join an open transaction, as Go's `WithTxV` has done since #1420. Active Record's `transaction` swallows an `ActiveRecord::Rollback` raised in a joined block and rolls nothing back. The Active Record helper therefore yields directly, so the rollback reaches the transaction's owner. Inside an open transaction, a `rollback_exception` raised in `Driver#transaction` now reaches the caller's enclosing transaction block, as a rollback raised in the caller's own code would. The SQLite batch insert uses the same helper. `RIVER_USE_LEGACY_SUBTRANSACTIONS=1` (or `true`) restores the savepoints, matching Go's temporary fallback of the same name. New specs run a scheduler pass with both kinds of conflict inside a caller's transaction. They assert the outcome that Go produces and that no savepoint opens. On Postgres, they also check that each changed row has the caller's transaction ID as its `xmin`. Specs that relied on River's savepoints now use one that the application owns. --- ruby/CHANGELOG.md | 3 + ruby/docs/README.md | 11 +- .../riverqueue-activerecord/lib/driver.rb | 19 ++- ruby/driver/riverqueue-sequel/lib/driver.rb | 14 +- ruby/lib/driver/runtime.rb | 149 +++++++++++++++--- ruby/sig/driver.rbs | 4 +- ruby/spec/client_driver_shared_examples.rb | 13 +- ruby/spec/driver_runtime_feature_spec.rb | 8 + ruby/spec/driver_runtime_shared_examples.rb | 130 +++++++++++++-- ruby/spec/driver_shared_examples.rb | 66 ++++++-- ruby/spec/support/transaction_helpers.rb | 82 ++++++++++ 11 files changed, 424 insertions(+), 75 deletions(-) create mode 100644 ruby/spec/support/transaction_helpers.rb diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index 91c982f1f..21ff2795d 100644 --- a/ruby/CHANGELOG.md +++ b/ruby/CHANGELOG.md @@ -16,10 +16,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- River operations called inside an open Active Record or Sequel transaction now run in that transaction and open no savepoint of their own. If an operation raises, roll back the transaction, or wrap the call in your own savepoint to recover and continue. A `driver.rollback_exception` raised inside `Driver#transaction` now reaches the enclosing transaction block, as a rollback raised in your own code would. Before, it rolled back only River's savepoint. Set `RIVER_USE_LEGACY_SUBTRANSACTIONS=1` (or `true`) to restore the savepoints. This temporary fallback will be removed in a future release. [PR #1544](https://github.com/riverqueue/river/pull/1544). - `work_end(job, error)` hooks now return the attempt's error, matching Go. Return the incoming error to preserve it, a replacement exception to change it, or `nil` to clear a failure. The return value feeds the next hook in registration order. Existing observing hooks must explicitly return `error`. End hooks are skipped if `work_begin` raises or the worker raises an exception outside `StandardError`. [PR #1520](https://github.com/riverqueue/river/pull/1520). ### Fixed +- The scheduler no longer opens a savepoint for each due job to catch unique key conflicts. A pass that made more than 64 jobs available overflowed the Postgres subtransaction cache, which can slow queries across the server. The scheduler now finds conflicts with one statement on Postgres and checks before each write on SQLite. Conflicting jobs are still discarded with `unique_key_conflict: scheduler_discarded` metadata. On Postgres, the scheduler now waits for a due job that another transaction has locked, instead of skipping it. [PR #1544](https://github.com/riverqueue/river/pull/1544). + - Default periodic jobs to their occurrence time, preserving constructor and argument-level schedules. [PR #1537](https://github.com/riverqueue/river/pull/1537). - Fail retries that omit a saved cursor-only step, while preserving their checkpoints for recovery. Jobs with an existing step marker retain their behavior, including legacy stale cursor entries. [PR #1534](https://github.com/riverqueue/river/pull/1534). diff --git a/ruby/docs/README.md b/ruby/docs/README.md index f3f969f0b..2b5467f39 100644 --- a/ruby/docs/README.md +++ b/ruby/docs/README.md @@ -184,8 +184,8 @@ For simple jobs, `River::JobArgsHash.new(:kind, hash)` avoids defining an argume ### [Transactional enqueueing](https://riverqueue.com/docs/transactional-enqueueing) -Insertion hooks and middleware execute inside the insertion transaction. If -they raise, their database writes and the enqueue roll back together. Postgres +Insertion hooks and middleware execute inside the insertion transaction, so +their database writes and the enqueue commit or roll back together. Postgres insert notifications are delivered only when the surrounding transaction commits. Inserts automatically join a transaction opened through the same Active Record connection or Sequel database object. A rollback also rolls back the job: @@ -206,6 +206,13 @@ end The equivalent works inside `ActiveRecord::Base.transaction` with the Active Record driver. +River calls inside an open transaction run in that transaction and open no +savepoint of their own. If a River call raises, it may leave partial changes, +so roll back the transaction. To recover from the error and continue the +transaction, wrap the call in your own savepoint, such as +`DB.transaction(savepoint: true)` or +`ApplicationRecord.transaction(requires_new: true)`. + ### Transactional completion Call `job.client.job_complete_tx(job)` inside a transaction to commit application diff --git a/ruby/driver/riverqueue-activerecord/lib/driver.rb b/ruby/driver/riverqueue-activerecord/lib/driver.rb index 9363a608c..ba4e44f10 100644 --- a/ruby/driver/riverqueue-activerecord/lib/driver.rb +++ b/ruby/driver/riverqueue-activerecord/lib/driver.rb @@ -110,10 +110,17 @@ def migration_connection end end - # Runs the block in a new Active Record transaction or savepoint and returns - # the block's result. + # Runs the block in an Active Record transaction and returns the block's + # result. Joins an open, joinable transaction instead of opening a + # savepoint. def transaction(&) - @connection_class.transaction(requires_new: true, &) + legacy = runtime_legacy_subtransactions? + # A joined Active Record block swallows ActiveRecord::Rollback without + # rolling anything back. Yield directly so that the transaction's owner + # receives it and rolls back, as Sequel does with Sequel::Rollback. + return yield if !legacy && @connection_class.connection.current_transaction.joinable? + + @connection_class.transaction(requires_new: legacy, &) end SQLITE_CONFLICT_WHERE = <<~SQL.chomp.freeze @@ -278,10 +285,6 @@ def transaction(&) @connection_class.connection.quote(value) end - private def runtime_unique_violation_class - ::ActiveRecord::RecordNotUnique - end - private def runtime_value(row, key) row[key.to_s] || row[key] end @@ -310,7 +313,7 @@ def transaction(&) # matching migration 007 and newer River databases. private def sqlite_job_insert_many(insert_params_many) nonces = sqlite_insert_nonces(insert_params_many) - @connection_class.transaction(requires_new: true) do + transaction do jobs = insert_params_many.zip(nonces).map { |param, nonce| sqlite_insert_params_to_hash(param, nonce) } inserted_nonces = nonces.to_h { |nonce| [nonce, true] } diff --git a/ruby/driver/riverqueue-sequel/lib/driver.rb b/ruby/driver/riverqueue-sequel/lib/driver.rb index 895022800..f550a85a5 100644 --- a/ruby/driver/riverqueue-sequel/lib/driver.rb +++ b/ruby/driver/riverqueue-sequel/lib/driver.rb @@ -89,11 +89,13 @@ def migration_connection end end - # Runs the block in a Sequel transaction or savepoint and returns the - # block's result. + # Runs the block in a Sequel transaction and returns the block's result. + # Joins an open transaction instead of opening a savepoint, unless that + # transaction uses Sequel's `auto_savepoint`. def transaction block_error = nil - @db.transaction(savepoint: true) do + options = runtime_legacy_subtransactions? ? {savepoint: true} : {} + @db.transaction(**options) do yield rescue ArgumentError => error block_error = error @@ -240,10 +242,6 @@ def transaction @db.literal(value) end - private def runtime_unique_violation_class - ::Sequel::UniqueConstraintViolation - end - private def runtime_value(row, key) row[key] || row[key.to_s] end @@ -272,7 +270,7 @@ def transaction # matching migration 007 and newer River databases. private def sqlite_job_insert_many(insert_params_array) nonces = sqlite_insert_nonces(insert_params_array) - @db.transaction(savepoint: true) do + transaction do jobs = insert_params_array.zip(nonces).map { |param, nonce| sqlite_insert_params_to_hash(param, nonce) } inserted_nonces = nonces.to_h { |nonce| [nonce, true] } diff --git a/ruby/lib/driver/runtime.rb b/ruby/lib/driver/runtime.rb index 6a7a8a7c4..39a46c0f0 100644 --- a/ruby/lib/driver/runtime.rb +++ b/ruby/lib/driver/runtime.rb @@ -275,35 +275,18 @@ def job_retry(id, now: Time.now.utc) job end + # Makes due scheduled and retryable jobs available. A job is discarded + # instead if another row already holds its unique key, or if an earlier + # job in the same pass shares that key. The queries find these conflicts + # before they write, so the pass never needs a savepoint to recover from a + # failed write. def job_schedule(now: Time.now.utc, max: 1_000) transaction do - queues = [] #: Array[String] - # Hold each selected row until its transition (including uniqueness - # conflict handling) finishes. A concurrent retry may change its due time. - lock = runtime_postgres? ? "FOR UPDATE SKIP LOCKED" : "" - rows = runtime_query_rows(<<~SQL) - SELECT id, queue FROM river_job - WHERE state IN ('retryable', 'scheduled') AND scheduled_at <= #{runtime_time(now)} - ORDER BY priority, scheduled_at, id - LIMIT #{Integer(max)} #{lock} - SQL - rows.each do |row| - id = runtime_value(row, :id).to_i - transaction do - runtime_execute("UPDATE river_job SET state = 'available' WHERE id = #{id} AND state IN ('retryable', 'scheduled')") - queues << runtime_value(row, :queue) - end - rescue runtime_unique_violation_class - runtime_execute(<<~SQL) - UPDATE river_job - SET state = 'discarded', finalized_at = #{runtime_time(now)}, - metadata = #{runtime_merge_metadata("unique_key_conflict" => "scheduler_discarded")} - WHERE id = #{id} - SQL - end - + queues = runtime_postgres? ? runtime_schedule_postgres(now, Integer(max)) : runtime_schedule_sqlite(now, Integer(max)) + # Notify each changed queue, including queues with only discarded + # jobs, as Go does. queues.uniq.each { |queue| runtime_notify("river_insert", queue: queue) } - rows.length + queues.length end end @@ -676,6 +659,12 @@ def time_precision runtime_postgres? ? "#{runtime_quote(encoded)}::jsonb" : "jsonb(#{runtime_quote(encoded)})" end + # Temporary fallback: when true, transaction helpers open a savepoint + # inside an open transaction instead of joining it. + private def runtime_legacy_subtransactions? + %w[1 true].include?(ENV.fetch("RIVER_USE_LEGACY_SUBTRANSACTIONS", "")) + end + private def runtime_merge_metadata(metadata) raise ArgumentError, "metadata must be a Hash" unless metadata.is_a?(Hash) @@ -777,6 +766,114 @@ def time_precision runtime_query_rows(sql).map { |row| runtime_value(row, :id).to_i } end + # Finds conflicts and changes the whole pass in one statement. + private def runtime_schedule_postgres(now, max) + now = runtime_time(now) + runtime_query_rows(<<~SQL).map { |row| runtime_value(row, :queue) } + WITH jobs_to_schedule AS ( + SELECT id, unique_key, unique_states, priority, scheduled_at + FROM river_job + WHERE state IN ('retryable', 'scheduled') + AND priority >= 0 + AND queue IS NOT NULL + AND scheduled_at <= #{now} + ORDER BY priority, scheduled_at, id + LIMIT #{max} + FOR UPDATE + ), + jobs_with_rownum AS ( + SELECT + *, + CASE + WHEN unique_key IS NOT NULL AND unique_states IS NOT NULL THEN + ROW_NUMBER() OVER (PARTITION BY unique_key ORDER BY priority, scheduled_at, id) + END AS row_num + FROM jobs_to_schedule + ), + unique_conflicts AS ( + SELECT river_job.unique_key + FROM river_job + JOIN jobs_with_rownum + ON river_job.unique_key = jobs_with_rownum.unique_key + AND river_job.id != jobs_with_rownum.id + WHERE river_job.unique_key IS NOT NULL + AND river_job.unique_states IS NOT NULL + AND river_job_state_in_bitmask(river_job.unique_states, river_job.state) + ), + job_updates AS ( + SELECT + job.id, + job.row_num IS NOT NULL AND (uc.unique_key IS NOT NULL OR job.row_num > 1) AS discard + FROM jobs_with_rownum job + LEFT JOIN unique_conflicts uc ON job.unique_key = uc.unique_key + ) + UPDATE river_job + SET state = CASE WHEN job_updates.discard THEN 'discarded'::river_job_state ELSE 'available'::river_job_state END, + finalized_at = CASE WHEN job_updates.discard THEN #{now} ELSE river_job.finalized_at END, + metadata = CASE WHEN job_updates.discard THEN river_job.metadata || '{"unique_key_conflict": "scheduler_discarded"}'::jsonb + ELSE river_job.metadata END + FROM job_updates + WHERE river_job.id = job_updates.id + RETURNING river_job.queue + SQL + end + + # Follows Go's SQLite driver: check each unique job in order, and make it + # available before checking the next one, so a later job with the same + # key sees it as a conflict. SQLite allows one writer at a time. Once this + # pass writes, other connections cannot write until it ends. If another + # connection writes first, this pass's next write fails instead of using + # stale rows. + private def runtime_schedule_sqlite(now, max) + rows = runtime_query_rows(<<~SQL) + SELECT id, queue, unique_key IS NOT NULL AS has_unique_key + FROM river_job + WHERE state IN ('retryable', 'scheduled') AND scheduled_at <= #{runtime_time(now)} + ORDER BY priority, scheduled_at, id + LIMIT #{max} + SQL + available = [] #: Array[Integer] + discarded = [] #: Array[Integer] + rows.each do |row| + id = runtime_value(row, :id).to_i + next available << id if runtime_value(row, :has_unique_key).to_i.zero? + + conflict = runtime_query_rows(<<~SQL).any? + SELECT 1 FROM river_job + WHERE id <> #{id} + AND unique_key = (SELECT job.unique_key FROM river_job AS job WHERE job.id = #{id}) + AND unique_states IS NOT NULL + AND CASE state + WHEN 'available' THEN unique_states & (1 << 0) + WHEN 'cancelled' THEN unique_states & (1 << 1) + WHEN 'completed' THEN unique_states & (1 << 2) + WHEN 'discarded' THEN unique_states & (1 << 3) + WHEN 'pending' THEN unique_states & (1 << 4) + WHEN 'retryable' THEN unique_states & (1 << 5) + WHEN 'running' THEN unique_states & (1 << 6) + WHEN 'scheduled' THEN unique_states & (1 << 7) + ELSE 0 + END >= 1 + LIMIT 1 + SQL + next discarded << id if conflict + + runtime_execute("UPDATE river_job SET state = 'available' WHERE id = #{id}") + end + + unless discarded.empty? + runtime_execute(<<~SQL) + UPDATE river_job + SET state = 'discarded', finalized_at = #{runtime_time(now)}, + metadata = #{runtime_merge_metadata("unique_key_conflict" => "scheduler_discarded")} + WHERE id IN (#{discarded.join(",")}) + SQL + end + runtime_execute("UPDATE river_job SET state = 'available' WHERE id IN (#{available.join(",")})") unless available.empty? + + rows.map { |row| runtime_value(row, :queue) } + end + private def runtime_state(value) runtime_postgres? ? "#{runtime_quote(value)}::river_job_state" : runtime_quote(value) end diff --git a/ruby/sig/driver.rbs b/ruby/sig/driver.rbs index bcc0b3760..7e0601398 100644 --- a/ruby/sig/driver.rbs +++ b/ruby/sig/driver.rbs @@ -177,6 +177,7 @@ module River def runtime_job_list: (JobListParams, ?for_delete: bool) -> Array[JobRow] def runtime_job_rows: (String) -> untyped def runtime_json: (untyped) -> untyped + def runtime_legacy_subtransactions?: () -> bool def runtime_merge_metadata: (untyped) -> untyped def runtime_metadata_equals: (untyped, untyped) -> untyped def runtime_nullable_time: (untyped) -> untyped @@ -193,10 +194,11 @@ module River def runtime_quote: (untyped) -> String def runtime_rescue_retry: (untyped, untyped, untyped, untyped, untyped) -> Time def runtime_returning_ids: (untyped) -> untyped + def runtime_schedule_postgres: (Time, Integer) -> Array[String] + def runtime_schedule_sqlite: (Time, Integer) -> Array[String] def runtime_state: (untyped) -> untyped def runtime_tag_contains: (untyped) -> untyped def runtime_time: (untyped) -> untyped - def runtime_unique_violation_class: () -> singleton(Exception) def runtime_update_value: (untyped, untyped) -> untyped def runtime_value: (untyped, untyped) -> untyped def transaction: () { () -> untyped } -> untyped diff --git a/ruby/spec/client_driver_shared_examples.rb b/ruby/spec/client_driver_shared_examples.rb index 674e09537..436e39936 100644 --- a/ruby/spec/client_driver_shared_examples.rb +++ b/ruby/spec/client_driver_shared_examples.rb @@ -207,7 +207,10 @@ end RSpec.shared_examples "SQL scheduling concurrency" do - it "skips locked jobs without overwriting a concurrent reschedule" do + # Like Go, the scheduler waits for a row that another transaction has + # locked. After the lock is released, Postgres rechecks the row, so a job + # rescheduled into the future stays scheduled. + it "waits for locked jobs without overwriting a concurrent reschedule" do client = River::Client.new(@driver) now = Time.now.utc first, second = [now - 2, now - 1].map do |scheduled_at| @@ -225,15 +228,19 @@ end Timeout.timeout(5) { locked.pop } - expect(@driver.job_schedule(now: now, max: 1)).to eq(1) - expect(@driver.job_get_by_id(second.id).state).to eq("available") + scheduler = Thread.new { @driver.job_schedule(now: now) } + expect(scheduler.join(0.5)).to be_nil release << true writer.value + + expect(scheduler.value).to eq(1) expect(@driver.job_get_by_id(first.id)).to have_attributes(state: "scheduled", scheduled_at: be_within(0.001).of(now + 60)) + expect(@driver.job_get_by_id(second.id).state).to eq("available") expect(@driver.job_schedule(now: now)).to eq(0) ensure release << true if release writer&.join + scheduler&.join end end diff --git a/ruby/spec/driver_runtime_feature_spec.rb b/ruby/spec/driver_runtime_feature_spec.rb index ceaa61ac1..8cfef94f7 100644 --- a/ruby/spec/driver_runtime_feature_spec.rb +++ b/ruby/spec/driver_runtime_feature_spec.rb @@ -21,6 +21,14 @@ it_behaves_like "driver timestamp precision" end +RSpec.describe "SQLite scheduler unique conflicts", database: :sqlite do + around(:each) { |example| sqlite_test_transaction(&example) } + + let(:driver) { River::Driver::Sequel.new(SQLITE_DB) } + + it_behaves_like "scheduler unique conflicts" +end + RSpec.describe "SQLite numeric metadata counters", database: :sqlite do around(:each) { |example| sqlite_test_transaction(&example) } diff --git a/ruby/spec/driver_runtime_shared_examples.rb b/ruby/spec/driver_runtime_shared_examples.rb index b0a74d6ac..785b8a906 100644 --- a/ruby/spec/driver_runtime_shared_examples.rb +++ b/ruby/spec/driver_runtime_shared_examples.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require "delegate" +require_relative "support/transaction_helpers" class DriverRuntimeArgs def initialize(value) @@ -93,9 +94,64 @@ def to_json = JSON.dump(value: @value) end end +RSpec.shared_examples "scheduler unique conflicts" do + let(:client) { River::Client.new(driver) } + let(:now) { Time.utc(2026, 1, 2, 3, 4, 5) } + + def insert_scheduler_job(value, **options) + client.insert(DriverRuntimeArgs.new(value), insert_opts: River::InsertOpts.new(state: River::JOB_STATE_AVAILABLE, **options)).job + end + + # Covers both kinds of scheduler conflict in one pass: a job that conflicts + # with an existing row, and two due jobs that share a unique key. Each + # example already runs in a caller's transaction, so the pass must reuse it. + it "schedules unique conflicts in the caller's transaction without savepoints" do + unique = River::UniqueOpts.new(by_args: true, by_state: %w[available pending running scheduled]) + conflicting = insert_scheduler_job(1, queue: "conflict_only", scheduled_at: now - 4, unique_opts: unique) + client.job_update(conflicting.id, state: River::JOB_STATE_RETRYABLE) + blocker = insert_scheduler_job(1, unique_opts: unique) + first, second = [now - 3, now - 2].map do |scheduled_at| + insert_scheduler_job(2, scheduled_at: scheduled_at, unique_opts: unique).tap do |job| + client.job_update(job.id, state: River::JOB_STATE_RETRYABLE) + end + end + plain = insert_scheduler_job(3, scheduled_at: now - 1, state: River::JOB_STATE_SCHEDULED) + + notified = [] #: Array[String] + driver.define_singleton_method(:runtime_notify) do |topic, payload| + notified << payload[:queue] if topic == "river_insert" + super(topic, payload) + end + + scheduled = nil + savepoints = savepoint_statements(driver) { scheduled = driver.job_schedule(now: now) } + + expect(scheduled).to eq(4) + expect(savepoints).to eq([]) + # Like Go, notify a queue even when its only changed job was discarded. + expect(notified).to contain_exactly("conflict_only", River::QUEUE_DEFAULT) + rows = [conflicting, blocker, first, second, plain].map { |job| driver.job_get_by_id(job.id) } + expect(rows.map(&:state)).to eq(%w[discarded available available discarded available]) + expect(rows.map { |row| row.metadata["unique_key_conflict"] }).to eq(["scheduler_discarded", nil, nil, "scheduler_discarded", nil]) + expect(rows.values_at(0, 3).map(&:finalized_at)).to all(be_within(0.001).of(now)) + expect(rows.values_at(1, 2, 4).map(&:finalized_at)).to all(be_nil) + + if driver.migration_backend == :postgresql + # A savepoint that writes gets its own transaction ID, so rows written in + # one would carry an xmin other than the caller's. + xmins = driver.send(:runtime_query_rows, <<~SQL).map { |row| driver.send(:runtime_value, row, :xid_matches) } + SELECT xmin = pg_current_xact_id()::xid AS xid_matches + FROM river_job WHERE id IN (#{[conflicting, first, second, plain].map(&:id).join(",")}) + SQL + expect(xmins).to eq([true] * 4) + end + end +end + RSpec.shared_examples "driver job state machine" do it_behaves_like "driver timestamp precision" it_behaves_like "overflowing metadata counters" + it_behaves_like "scheduler unique conflicts" let(:client) { River::Client.new(driver) } let(:now) { Time.utc(2026, 1, 2, 3, 4, 5) } @@ -144,10 +200,12 @@ def to_json = JSON.dump(value: @value) hooked_client = River::Client.new(driver, config: River::Config.new(plugins: [plugin])) expect do - if operation == :insert - hooked_client.insert(DriverRuntimeArgs.new(1)) - else - hooked_client.insert_many([DriverRuntimeArgs.new(1), DriverRuntimeArgs.new(2)]) + savepoint(driver) do + if operation == :insert + hooked_client.insert(DriverRuntimeArgs.new(1)) + else + hooked_client.insert_many([DriverRuntimeArgs.new(1), DriverRuntimeArgs.new(2)]) + end end end.to raise_error(ArgumentError, 'invalid insertion state: "running"; must be available, pending, or scheduled') expect(plain_client.job_list.jobs).to be_empty @@ -192,6 +250,23 @@ def to_json = JSON.dump(value: @value) it "preserves application argument errors and rolls back their transaction writes" do failure = ArgumentError.new("invalid application argument") + expect do + savepoint(driver) do + driver.transaction do + client.insert(DriverRuntimeArgs.new(1)) + raise failure + end + end + end.to raise_error { |error| expect(error).to equal(failure) } + expect(client.job_list.jobs).to be_empty + expect(client.insert(DriverRuntimeArgs.new(2)).job.state).to eq("available") + end + + it "preserves application argument errors from a legacy savepoint and rolls back its writes" do + failure = ArgumentError.new("invalid application argument") + previous = ENV["RIVER_USE_LEGACY_SUBTRANSACTIONS"] + ENV["RIVER_USE_LEGACY_SUBTRANSACTIONS"] = "1" + expect do driver.transaction do client.insert(DriverRuntimeArgs.new(1)) @@ -200,6 +275,8 @@ def to_json = JSON.dump(value: @value) end.to raise_error { |error| expect(error).to equal(failure) } expect(client.job_list.jobs).to be_empty expect(client.insert(DriverRuntimeArgs.new(2)).job.state).to eq("available") + ensure + ENV["RIVER_USE_LEGACY_SUBTRANSACTIONS"] = previous end it "completes running jobs, merges metadata, and ignores missing or finished jobs" do @@ -218,7 +295,7 @@ def to_json = JSON.dump(value: @value) driver.job_metadata_merge(row.id, "cancel_attempted_at" => nil) expect(driver.job_complete(id: row.id, finalized_at: now)).to eq(:cancelled) expect(driver.job_get_by_id(row.id).state).to eq("running") - driver.transaction do + savepoint(driver) do expect(driver.job_complete(id: row.id, finalized_at: now)).to eq(:cancelled) raise driver.rollback_exception end @@ -338,7 +415,7 @@ def work(_job) it "rolls back transactional completion" do row = insert_job(1, scheduled_at: now - 1) driver.job_claim(id: row.id, attempted_by: "test", now: now) - driver.transaction do + savepoint(driver) do expect(driver.job_complete(id: row.id, finalized_at: now).state).to eq("completed") raise driver.rollback_exception end @@ -353,7 +430,7 @@ def work(_job) driver.job_metadata_merge(null_marker.id, "cancel_attempted_at" => nil) expect(driver.job_get_cancelled_ids([])).to eq([]) expect(driver.job_get_cancelled_ids([cancelled.id, unmarked.id, null_marker.id, -1, cancelled.id])).to contain_exactly(cancelled.id, null_marker.id) - driver.transaction do + savepoint(driver) do driver.job_metadata_merge(unmarked.id, "cancel_attempted_at" => "now") expect(driver.job_get_cancelled_ids([unmarked.id])).to eq([unmarked.id]) raise driver.rollback_exception @@ -939,6 +1016,37 @@ def insert_job(value, **options) expect(discarded.metadata.to_h).to include("unique_key_conflict" => "scheduler_discarded") end + it "runs transactional operations in the caller's transaction without savepoints" do + queue = "no_savepoints" + driver.queue_upsert(queue, now: now) + retried = insert_job(1, queue: queue, scheduled_at: now - 2) + driver.job_cancel(retried.id, now: now) + stuck = insert_job(2, queue: queue, scheduled_at: now - 120) + driver.job_claim(id: stuck.id, attempted_by: "worker", now: now - 120) + + savepoints = savepoint_statements(driver) do + driver.transaction do + client.insert(DriverRuntimeArgs.new(3), queue: queue) + client.insert_many([4, 5].map { |value| River::InsertManyParams.new(DriverRuntimeArgs.new(value), insert_opts: River::InsertOpts.new(queue: queue)) }) + end + claimed = driver.job_get_available(attempted_by: "worker", max: 1, queue: queue).first + driver.job_metadata_merge(claimed.id, {"merged" => true}) + expect(client.job_complete_tx(River::Job.new(client, claimed)).state).to eq("completed") + driver.job_cancel(insert_job(6, queue: queue).id, now: now) + driver.job_retry(retried.id, now: now) + driver.job_rescue_stuck(horizon: now - 60, now: now, retry_policy: River::DefaultClientRetryPolicy.new) + driver.job_delete_many(River::JobListParams.new(kinds: ["driver_runtime"], queues: [queue], states: [River::JOB_STATE_CANCELLED])) + driver.queue_pause(queue, now: now) + driver.queue_resume(queue, now: now) + driver.queue_update(queue, metadata: {"updated" => true}, now: now) + driver.leader_acquire("no-savepoints", now: now) + driver.leader_release("no-savepoints") + driver.request_resign + end + + expect(savepoints).to eq([]) + end + it "continues past ineligible rescue candidates without consuming the limit" do jobs = 3.times.map do |index| inserted = insert_job(index, scheduled_at: now - 120) @@ -1239,13 +1347,13 @@ def insert_job(value, **options) driver.queue_pause("paused", now: now) driver.leader_acquire("leader") driver.define_singleton_method(:runtime_notify) { |*| raise "notification failed" } - expect { driver.queue_pause("priority") }.to raise_error("notification failed") + expect { savepoint(driver) { driver.queue_pause("priority") } }.to raise_error("notification failed") expect(driver.queue_get("priority").paused_at).to be_nil - expect { driver.queue_resume("paused") }.to raise_error("notification failed") + expect { savepoint(driver) { driver.queue_resume("paused") } }.to raise_error("notification failed") expect(driver.queue_get("paused").paused_at).to be_within(0.001).of(now) - expect { driver.queue_update("priority", metadata: {"changed" => true}) }.to raise_error("notification failed") + expect { savepoint(driver) { driver.queue_update("priority", metadata: {"changed" => true}) } }.to raise_error("notification failed") expect(driver.queue_get("priority").metadata).to eq({}) - expect { driver.leader_release("leader") }.to raise_error("notification failed") + expect { savepoint(driver) { driver.leader_release("leader") } }.to raise_error("notification failed") expect(driver.leader_renew("leader")).to be true end diff --git a/ruby/spec/driver_shared_examples.rb b/ruby/spec/driver_shared_examples.rb index 399da71ab..893297bc5 100644 --- a/ruby/spec/driver_shared_examples.rb +++ b/ruby/spec/driver_shared_examples.rb @@ -111,7 +111,7 @@ class SimpleArgsWithInsertOpts < SimpleArgs end %i[insert_begin insert_end insert_many].each do |hook| - it "rolls back #{hook} writes and the enqueue without aborting the caller's transaction" do + it "rolls back #{hook} writes and the enqueue in an application savepoint" do plugin = Object.new plain_client = client failure = RuntimeError.new("hook failed") @@ -124,7 +124,8 @@ class SimpleArgsWithInsertOpts < SimpleArgs driver.transaction do kept = plain_client.insert(SimpleArgs.new(job_num: 1)).job - expect { hooked_client.insert(SimpleArgs.new(job_num: 2)) }.to raise_error { |error| expect(error).to equal(failure) } + expect { savepoint(driver) { hooked_client.insert(SimpleArgs.new(job_num: 2)) } } + .to raise_error { |error| expect(error).to equal(failure) } expect(plain_client.job_list.jobs.map(&:id)).to eq([kept.id]) expect(plain_client.insert(SimpleArgs.new(job_num: 3)).job.args).to eq("job_num" => 3) end @@ -150,7 +151,7 @@ class SimpleArgsWithInsertOpts < SimpleArgs job = client.insert(SimpleArgs.new(job_num: 1)).job driver.define_singleton_method(:runtime_notify) { |*_args| raise "notification failed" } - expect { client.job_cancel(job.id) }.to raise_error("notification failed") + expect { savepoint(driver) { client.job_cancel(job.id) } }.to raise_error("notification failed") expect(client.job_get(job.id)).to have_attributes(state: "available", metadata: job.metadata) end @@ -184,7 +185,6 @@ class SimpleArgsWithInsertOpts < SimpleArgs expect(driver.send(:runtime_job_list_without_params)).to be_an(Array) expect(driver.send(:runtime_postgres?)).to satisfy { |value| value == true || value == false } expect(driver.send(:runtime_quote, "value")).to be_a(String) - expect(driver.send(:runtime_unique_violation_class)).to be <= StandardError expect(driver.send(:runtime_value, {"id" => 1}, :id)).to eq(1) end @@ -194,7 +194,7 @@ class SimpleArgsWithInsertOpts < SimpleArgs options = River::InsertOpts.new(unique_opts: River::UniqueOpts.new(by_args: true)) batch = [2, 1, 1].map { |number| River::InsertManyParams.new(SimpleArgs.new(job_num: number), insert_opts: options) } - expect { client.insert_many(batch) }.to raise_error(StandardError) + expect { savepoint(driver) { client.insert_many(batch) } }.to raise_error(StandardError) expect(client.job_list.jobs.map(&:id)).to eq([kept.id]) end @@ -408,7 +408,7 @@ class SimpleArgsWithInsertOpts < SimpleArgs it "inserts in a transaction" do insert_res = nil - driver.transaction do + savepoint(driver) do insert_res = client.insert(SimpleArgs.new(job_num: 1)) job = driver.job_get_by_id(insert_res.job.id) @@ -529,7 +529,7 @@ class SimpleArgsWithInsertOpts < SimpleArgs it "inserts multiple jobs in a transaction" do jobs = nil - driver.transaction do + savepoint(driver) do inserted = client.insert_many([ SimpleArgs.new(job_num: 1), SimpleArgs.new(job_num: 2) @@ -601,23 +601,57 @@ class SimpleArgsWithInsertOpts < SimpleArgs end describe "#transaction" do - it "runs block in a transaction" do + it "joins the caller's transaction without a savepoint" do + # Until the outer transaction runs a query, Active Record restarts it + # instead of opening a savepoint, so run one first to make an unwanted + # savepoint visible. + client.insert(SimpleArgs.new(job_num: 0)) insert_res = nil - driver.transaction do - insert_res = client.insert(SimpleArgs.new(job_num: 1)) + savepoints = savepoint_statements(driver) do + driver.transaction do + insert_res = client.insert(SimpleArgs.new(job_num: 1)) + end + end - job = driver.job_get_by_id(insert_res.job.id) + expect(savepoints).to eq([]) + expect(driver.job_get_by_id(insert_res.job.id)).to_not be_nil + end - expect(job).to_not be_nil + %w[1 true].each do |value| + it "opens a savepoint when RIVER_USE_LEGACY_SUBTRANSACTIONS is #{value}" do + previous = ENV["RIVER_USE_LEGACY_SUBTRANSACTIONS"] + ENV["RIVER_USE_LEGACY_SUBTRANSACTIONS"] = value + insert_res = kept = nil + + savepoints = savepoint_statements(driver) do + kept = client.insert(SimpleArgs.new(job_num: 1)).job + driver.transaction do + insert_res = client.insert(SimpleArgs.new(job_num: 2)) + raise driver.rollback_exception + end + end - raise driver.rollback_exception + expect(savepoints.grep(/\ASAVEPOINT/i)).not_to be_empty + expect(driver.job_get_by_id(insert_res.job.id)).to be_nil + expect(driver.job_get_by_id(kept.id)).to_not be_nil + ensure + ENV["RIVER_USE_LEGACY_SUBTRANSACTIONS"] = previous end + end - # Not present because the job was rolled back. - job = driver.job_get_by_id(insert_res.job.id) + it "passes a rollback to the caller's transaction" do + insert_res = nil - expect(job).to be_nil + savepoint(driver) do + driver.transaction do + insert_res = client.insert(SimpleArgs.new(job_num: 1)) + raise driver.rollback_exception + end + raise "unreachable" + end + + expect(driver.job_get_by_id(insert_res.job.id)).to be_nil end end end diff --git a/ruby/spec/support/transaction_helpers.rb b/ruby/spec/support/transaction_helpers.rb new file mode 100644 index 000000000..813fed6d9 --- /dev/null +++ b/ruby/spec/support/transaction_helpers.rb @@ -0,0 +1,82 @@ +# frozen_string_literal: true + +require "logger" +require "stringio" + +# Helpers for examples that run inside a caller's transaction, which River's +# drivers join instead of opening a savepoint of their own. +module TransactionHelpers + # Runs the block in an application-owned savepoint, the way an application + # recovers from a failed River call and keeps its transaction. Re-raises + # the block's error after rolling back. Raising it outside the savepoint + # keeps Sequel's SQLite adapter from wrapping an ArgumentError. + def savepoint(driver) + error = nil + options = driver.respond_to?(:connection_class) ? [driver.connection_class, {requires_new: true}] : [driver.instance_variable_get(:@db), {savepoint: true}] + result = options[0].transaction(**options[1]) do + yield + rescue driver.rollback_exception + raise + rescue => block_error + error = block_error + raise driver.rollback_exception + end + raise error if error + + result + end + + # Returns the savepoints that the driver's adapter creates, releases, or + # rolls back to while the block runs. + def savepoint_statements(driver, &) + return sql_statements(driver, &).grep(/\A\s*(RELEASE\s+|ROLLBACK\s+TO\s+)?SAVEPOINT\b/i) unless driver.respond_to?(:connection_class) + + # Record Active Record's savepoint calls directly. Lazily created + # savepoints are not always logged before the block returns. + connection = driver.connection_class.connection + connection.singleton_class.prepend(ActiveRecordSavepointRecorder) unless connection.singleton_class <= ActiveRecordSavepointRecorder + statements = [] #: Array[String] + connection.river_savepoint_statements = statements + begin + yield + ensure + connection.river_savepoint_statements = nil + end + statements + end + + # Returns the SQL that the driver's adapter sends while the block runs. + def sql_statements(driver) + db = driver.instance_variable_get(:@db) + output = StringIO.new + logger = Logger.new(output, formatter: ->(_severity, _time, _program, message) { "#{message}\n" }) + db.loggers << logger + begin + yield + ensure + db.loggers.delete(logger) + end + output.string.lines.map { |line| line.sub(/\A\([\d.]+s\) /, "").strip } + end + + module ActiveRecordSavepointRecorder + attr_accessor :river_savepoint_statements + + def create_savepoint(name = current_savepoint_name) + river_savepoint_statements&.push("SAVEPOINT #{name}") + super + end + + def exec_rollback_to_savepoint(name = current_savepoint_name) + river_savepoint_statements&.push("ROLLBACK TO SAVEPOINT #{name}") + super + end + + def release_savepoint(name = current_savepoint_name) + river_savepoint_statements&.push("RELEASE SAVEPOINT #{name}") + super + end + end +end + +RSpec.configure { |config| config.include TransactionHelpers }