From 97ef544203204cd61680ad537463147f3ad286bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Galisteo?= Date: Thu, 15 Oct 2026 23:38:03 +0200 Subject: [PATCH] Let jobs acquire their concurrency lock when dispatched When Solid Queue shares the app's connection and a concurrency-limited job is enqueued inside a transaction, acquiring the semaphore happens inside that transaction: Semaphore#wait selects the row FOR UPDATE, or inserts it, and the lock lasts until the caller's transaction ends. Until then, any other enqueue with the same key waits, and so does the worker signalling the semaphore after running a job with that key. Two transactions that enqueue jobs for two keys in opposite order deadlock, which surfaces as SolidQueue::Job::EnqueueError and aborts the caller's transaction. A job enqueued with a wait that isn't due yet when it's inserted doesn't have the problem: it's scheduled, and the dispatcher acquires its semaphore later in its own transaction. This makes that available on purpose, without a delay. A new option says when a job acquires its concurrency lock: - :on_enqueue, the default and the current behaviour. - :on_dispatch, enqueuing the job as a scheduled job that is already due, for the dispatcher to acquire the lock once it's committed. - :outside_transaction, :on_dispatch when there's an open transaction on Solid Queue's connection, :on_enqueue otherwise. It can be set globally (config.solid_queue.concurrency_lock_acquisition), per job (limits_concurrency acquire:) or per enqueue (set(acquire:)), the most specific one winning. The job row is still inserted in the caller's transaction, so it's still committed or rolled back with it. Whether there's an open transaction is checked before the job's own insert, which opens one, and only joinable transactions count, so the test fixtures' transaction doesn't. Jobs without concurrency controls have no lock to defer, so they skip the transaction check. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Lceyt66zciMB7d4Yv7UgDd --- README.md | 33 +++- app/models/solid_queue/job.rb | 33 +++- app/models/solid_queue/job/executable.rb | 21 ++- lib/active_job/concurrency_controls.rb | 22 ++- lib/solid_queue.rb | 18 ++ .../concurrency_lock_acquisition_test.rb | 125 ++++++++++++++ test/models/solid_queue/job_test.rb | 155 ++++++++++++++++++ 7 files changed, 399 insertions(+), 8 deletions(-) create mode 100644 test/integration/concurrency_lock_acquisition_test.rb diff --git a/README.md b/README.md index 4367c6b07..e6e8447b4 100644 --- a/README.md +++ b/README.md @@ -453,6 +453,7 @@ There are several settings that control how Solid Queue works that you can set a - `preserve_finished_jobs`: whether to keep finished jobs in the `solid_queue_jobs` table—defaults to `true`. - `clear_finished_jobs_after`: period to keep finished jobs around, in case `preserve_finished_jobs` is true — defaults to 1 day. When installing Solid Queue, [a recurring job](#recurring-tasks) is automatically configured to clear finished jobs every hour on the 12th minute in batches. You can edit the `recurring.yml` configuration to change this as you see fit. - `default_concurrency_control_period`: the value to be used as the default for the `duration` parameter in [concurrency controls](#concurrency-controls). It defaults to 3 minutes. +- `concurrency_lock_acquisition`: when jobs with [concurrency controls](#concurrency-controls) acquire their concurrency lock, unless the job or the enqueue says otherwise: `:on_enqueue` (the default), `:on_dispatch` or `:outside_transaction`. See [acquiring the concurrency lock when dispatched](#acquiring-the-concurrency-lock-when-dispatched). ### Validating the configuration @@ -533,7 +534,7 @@ Solid Queue extends Active Job with concurrency controls, that allows you to lim ```ruby class MyJob < ApplicationJob - limits_concurrency to: max_concurrent_executions, key: ->(arg1, arg2, *) { ... }, duration: max_interval_to_guarantee_concurrency_limit, group: concurrency_group, on_conflict: on_conflict_behaviour + limits_concurrency to: max_concurrent_executions, key: ->(arg1, arg2, *) { ... }, duration: max_interval_to_guarantee_concurrency_limit, group: concurrency_group, on_conflict: on_conflict_behaviour, acquire: lock_acquisition # ... ``` @@ -544,6 +545,7 @@ class MyJob < ApplicationJob - `on_conflict` controls behaviour when enqueuing a job that conflicts with the concurrency limits configured. It can be set to one of the following: - (default) `:block`: the job is blocked and is dispatched when another job completes and unblocks it, or when the duration expires. - `:discard`: the job is discarded. When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded until the interval defined by `duration` has elapsed. +- `acquire` controls when the job acquires its concurrency lock: `:on_enqueue`, `:on_dispatch` or `:outside_transaction`. It defaults to `SolidQueue.concurrency_lock_acquisition`, which is `:on_enqueue` unless you configure it. See [acquiring the concurrency lock when dispatched](#acquiring-the-concurrency-lock-when-dispatched). When a job includes these controls, we'll ensure that, at most, the number of jobs (indicated as `to`) that yield the same `key` will be performed concurrently, and this guarantee will last for `duration` for each job enqueued. Note that there's no guarantee about _the order of execution_, only about jobs being performed at the same time (overlapping). @@ -617,6 +619,35 @@ The 3 jobs will go into the scheduled queue and will wait there until they're du Normally scheduled jobs are enqueued in batches, but with concurrency controls, jobs need to be enqueued one by one. This has an impact on performance, similarly to the impact of concurrency controls in bulk enqueuing. Read below for more details. We generally advise against mixing concurrency controls with waiting/scheduling in the future. +### Acquiring the concurrency lock when dispatched + +By default, a job that is due acquires its concurrency lock as soon as it's enqueued. When Solid Queue uses the same database connection as your app (no `connects_to`) and the job is enqueued inside a transaction, without deferring it until the transaction commits (`enqueue_after_transaction_commit` disabled), that means acquiring it inside your transaction: the semaphore's row stays locked until your transaction commits or rolls back. Until then, any other enqueue of a job with the same concurrency key waits, and so does the worker that finishes a job with that key and tries to signal the semaphore. Two transactions that enqueue jobs with two different keys in opposite order can deadlock, which surfaces as a `SolidQueue::Job::EnqueueError` and aborts the transaction. + +To avoid this, a job can acquire its concurrency lock when it's dispatched instead. It's then enqueued as a scheduled job that is already due, and the dispatcher acquires the lock in its own transaction once the job is committed, and moves the job to ready or blocked from there. The job itself is still enqueued inside your transaction, so it's still committed or rolled back with it. + +There are three possible values: + +- `:on_enqueue`, the default, acquires the lock when the job is enqueued. +- `:on_dispatch` always leaves acquiring the lock to the dispatcher. +- `:outside_transaction` acquires the lock when the job is enqueued, unless it's enqueued inside an open transaction on Solid Queue's connection, in which case it leaves it to the dispatcher. Jobs enqueued outside a transaction aren't delayed at all, and if Solid Queue has its own database, nothing changes. + +You can choose one globally, for a job, or for a single enqueue, with the most specific one taking precedence: + +```ruby +# config/application.rb or an environment file +config.solid_queue.concurrency_lock_acquisition = :outside_transaction + +class DeliverAnnouncementToContactJob < ApplicationJob + limits_concurrency to: 1, key: ->(contact) { contact.account }, acquire: :on_dispatch +end + +DeliverAnnouncementToContactJob.set(acquire: :on_enqueue).perform_later(contact) +``` + +A job that leaves the lock to the dispatcher waits for the dispatcher's next poll before it can be ready, up to its `polling_interval`, and goes through the scheduled jobs path, one by one, as described in [scheduled jobs](#scheduled-jobs). With `on_conflict: :discard`, a conflicting job is discarded when it's dispatched rather than when it's enqueued, so the enqueue itself succeeds, as it does for jobs scheduled in the future. + +Bear in mind that these jobs depend on the dispatcher to run at all, like jobs scheduled in the future do: without a dispatcher running, they stay scheduled. + ### Performance considerations Concurrency controls introduce significant overhead (blocked executions need to be created and promoted to ready, semaphores need to be created and updated) so you should consider carefully whether you need them. For throttling purposes, where you plan to have `limit` significantly larger than 1, we encourage relying on a limited number of workers per queue instead. For example: diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb index cb8f25bee..b3a3796c2 100644 --- a/app/models/solid_queue/job.rb +++ b/app/models/solid_queue/job.rb @@ -27,9 +27,12 @@ def enqueue_all(active_jobs) job.batch_id = current_batch_id || job.batch_id end active_jobs_by_job_id = active_jobs.index_by(&:job_id) + deferring_lock = active_jobs.select { |job| defer_concurrency_lock_for?(job) }.to_set(&:job_id) transaction do jobs = create_all_from_active_jobs(active_jobs) + jobs.each { |job| job.defer_concurrency_lock = deferring_lock.include?(job.active_job_id) } + prepare_all_for_execution(jobs).tap do |enqueued_jobs| enqueued_jobs.each do |enqueued_job| active_jobs_by_job_id[enqueued_job.active_job_id].provider_job_id = enqueued_job.id @@ -44,7 +47,7 @@ def enqueue_all(active_jobs) def enqueue(active_job, scheduled_at: Time.current) active_job.scheduled_at = scheduled_at - create_from_active_job(active_job).tap do |enqueued_job| + create_from_active_job(active_job, defer_concurrency_lock: defer_concurrency_lock_for?(active_job)).tap do |enqueued_job| active_job.provider_job_id = enqueued_job.id if enqueued_job.persisted? active_job.successfully_enqueued = enqueued_job.persisted? end @@ -54,8 +57,10 @@ def enqueue(active_job, scheduled_at: Time.current) DEFAULT_PRIORITY = 0 DEFAULT_QUEUE_NAME = "default" - def create_from_active_job(active_job) - create!(**attributes_from_active_job(active_job)) + def create_from_active_job(active_job, defer_concurrency_lock: false) + create!(**attributes_from_active_job(active_job)) do |job| + job.defer_concurrency_lock = defer_concurrency_lock + end rescue ActiveRecord::ActiveRecordError => e enqueue_error = EnqueueError.new("#{e.class.name}: #{e.message}").tap do |error| error.set_backtrace e.backtrace @@ -63,6 +68,28 @@ def create_from_active_job(active_job) raise enqueue_error end + # Whether the job should acquire its concurrency lock when it's dispatched + # rather than right away. Decided before the job's own insert opens a + # transaction, so that one doesn't count as the caller's. + def defer_concurrency_lock_for?(active_job) + return false unless active_job.class.concurrency_key + + case active_job.concurrency_lock_acquisition + when :on_dispatch then true + when :outside_transaction then inside_transaction? + else false + end + end + + # A transaction the caller opened on Solid Queue's connection, which would + # hold the semaphore's row lock until it ends. The test fixtures' + # transaction isn't joinable, so it doesn't count. + def inside_transaction? + connection_pool.with_connection do |connection| + connection.transaction_open? && connection.current_transaction.joinable? + end + end + def create_all_from_active_jobs(active_jobs) job_rows = active_jobs.map { |job| attributes_from_active_job(job) } insert_all(job_rows) diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb index 75a5d2111..937d01d52 100644 --- a/app/models/solid_queue/job/executable.rb +++ b/app/models/solid_queue/job/executable.rb @@ -13,6 +13,10 @@ module Executable after_create :prepare_for_execution + # Set when enqueueing, for a job that acquires its concurrency lock when + # it's dispatched instead of right away. See SolidQueue.concurrency_lock_acquisition. + attr_writer :defer_concurrency_lock + scope :finished, -> { where.not(finished_at: nil) } end @@ -21,8 +25,8 @@ def prepare_all_for_execution(jobs) # Track before dispatch so conflict-discarded jobs count like single enqueues. batch_all(jobs) - due, not_yet_due = jobs.partition(&:due?) - dispatch_all(due) + schedule_all(not_yet_due) + dispatchable, schedulable = jobs.partition(&:dispatchable_on_enqueue?) + dispatch_all(dispatchable) + schedule_all(schedulable) end def dispatch_all(jobs) @@ -62,12 +66,23 @@ def dispatched_and_blocked(jobs_by_id) end def prepare_for_execution - if due? then dispatch + if dispatchable_on_enqueue? then dispatch else schedule end end + # A job that defers its concurrency lock is scheduled even when it's due: + # the dispatcher acquires the lock once the job is committed, outside the + # transaction that enqueued it. + def dispatchable_on_enqueue? + due? && !defer_concurrency_lock? + end + + def defer_concurrency_lock? + @defer_concurrency_lock.present? && concurrency_limited? + end + def dispatch if acquire_concurrency_lock then ready else diff --git a/lib/active_job/concurrency_controls.rb b/lib/active_job/concurrency_controls.rb index 47b0f5eea..eb7cc18f2 100644 --- a/lib/active_job/concurrency_controls.rb +++ b/lib/active_job/concurrency_controls.rb @@ -14,18 +14,32 @@ module ConcurrencyControls class_attribute :concurrency_limit class_attribute :concurrency_duration, default: SolidQueue.default_concurrency_control_period class_attribute :concurrency_on_conflict, default: :block + class_attribute :concurrency_acquisition, instance_accessor: false end class_methods do - def limits_concurrency(key:, to: 1, group: DEFAULT_CONCURRENCY_GROUP, duration: SolidQueue.default_concurrency_control_period, on_conflict: :block) + def limits_concurrency(key:, to: 1, group: DEFAULT_CONCURRENCY_GROUP, duration: SolidQueue.default_concurrency_control_period, on_conflict: :block, acquire: nil) self.concurrency_key = key self.concurrency_limit = to self.concurrency_group = group self.concurrency_duration = duration self.concurrency_on_conflict = on_conflict.presence_in(CONCURRENCY_ON_CONFLICT_BEHAVIOUR) || :block + self.concurrency_acquisition = acquire && SolidQueue.validate_concurrency_lock_acquisition!(acquire) end end + # Accepts +acquire+ on top of Active Job's own options, to choose when this + # particular enqueue acquires the concurrency lock: + # + # MyJob.set(acquire: :on_dispatch).perform_later(record) + def set(options = {}) + if options.key?(:acquire) + @concurrency_acquisition = options[:acquire] && SolidQueue.validate_concurrency_lock_acquisition!(options[:acquire]) + end + + super + end + def concurrency_key if self.class.concurrency_key param = compute_concurrency_parameter(self.class.concurrency_key) @@ -43,6 +57,12 @@ def concurrency_limited? concurrency_key.present? end + # When this job acquires its concurrency lock: set for this enqueue, for the + # job class, or globally, in that order. + def concurrency_lock_acquisition + @concurrency_acquisition || self.class.concurrency_acquisition || SolidQueue.concurrency_lock_acquisition + end + private def concurrency_group compute_concurrency_parameter(self.class.concurrency_group) diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb index fbc2f5e65..851a92b16 100644 --- a/lib/solid_queue.rb +++ b/lib/solid_queue.rb @@ -43,6 +43,24 @@ module SolidQueue mattr_accessor :clear_finished_jobs_after, default: 1.day mattr_accessor :default_concurrency_control_period, default: 3.minutes + # When a concurrency-limited job acquires its semaphore: + # - :on_enqueue, as soon as it's enqueued. + # - :on_dispatch, when the dispatcher picks it up, after enqueuing it as a scheduled job. + # - :outside_transaction, :on_enqueue unless the job is enqueued inside an open + # transaction on Solid Queue's connection, :on_dispatch otherwise. + CONCURRENCY_LOCK_ACQUISITIONS = %i[ on_enqueue on_dispatch outside_transaction ] + + mattr_reader :concurrency_lock_acquisition, default: :on_enqueue + + def concurrency_lock_acquisition=(value) + @@concurrency_lock_acquisition = validate_concurrency_lock_acquisition!(value) + end + + def validate_concurrency_lock_acquisition!(value) + value.try(:to_sym).presence_in(CONCURRENCY_LOCK_ACQUISITIONS) or + raise ArgumentError, "Unknown concurrency lock acquisition #{value.inspect}, expected one of #{CONCURRENCY_LOCK_ACQUISITIONS.inspect}" + end + mattr_reader :time_zone def time_zone=(zone) diff --git a/test/integration/concurrency_lock_acquisition_test.rb b/test/integration/concurrency_lock_acquisition_test.rb new file mode 100644 index 000000000..94bd46ffe --- /dev/null +++ b/test/integration/concurrency_lock_acquisition_test.rb @@ -0,0 +1,125 @@ +# frozen_string_literal: true + +require "test_helper" + +class ConcurrencyLockAcquisitionTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + class NonOverlappingJob < ApplicationJob + # Enqueued inside the transaction, which is what this is about. + # Rails 7.2 takes :never rather than false. + self.enqueue_after_transaction_commit = Rails.version.start_with?("7.2") ? :never : false if respond_to?(:enqueue_after_transaction_commit=) + + limits_concurrency key: ->(key) { key } + + def perform(key) + end + end + + setup do + # SQLite allows a single writer at a time, so any enqueue waits for any open + # write transaction, concurrency controls or not. + skip "SQLite serializes all writes" if SolidQueue::Record.connection_pool.db_config.adapter == "sqlite3" + end + + test "a job enqueued inside a transaction holds its concurrency lock until the transaction ends" do + assert_not completes_while_transaction_is_open?(acquire: :on_enqueue) { + NonOverlappingJob.perform_later("shared") + } + end + + test "a job leaving its concurrency lock to the dispatcher doesn't hold it in the transaction" do + assert completes_while_transaction_is_open?(acquire: :outside_transaction) { + NonOverlappingJob.perform_later("shared") + } + end + + test "a worker signalling the semaphore waits for a transaction whose enqueue was blocked" do + running = enqueue_running_job + + assert_not completes_while_transaction_is_open?(acquire: :on_enqueue) { + SolidQueue::Semaphore.signal(running) + } + end + + test "a worker signalling the semaphore doesn't wait for a transaction that left the lock to the dispatcher" do + running = enqueue_running_job + + assert completes_while_transaction_is_open?(acquire: :outside_transaction) { + SolidQueue::Semaphore.signal(running) + } + end + + test "transactions enqueueing jobs for two keys in opposite order deadlock" do + errors = enqueue_in_opposite_orders(acquire: :on_enqueue) + + assert_equal 1, errors.size + assert_kind_of SolidQueue::Job::EnqueueError, errors.first + assert_kind_of ActiveRecord::Deadlocked, errors.first.cause + end + + test "transactions leaving the lock to the dispatcher enqueue jobs for two keys in opposite order" do + assert_empty enqueue_in_opposite_orders(acquire: :outside_transaction) + end + + private + # Enqueues a job with `acquire` inside a transaction on Solid Queue's + # connection, and keeps that transaction open while the block runs on + # another connection. Answers whether the block completed before the + # transaction ended. + def completes_while_transaction_is_open?(acquire:, &block) + enqueued = Concurrent::Event.new + done = Concurrent::Event.new + + holder = Thread.new do + SolidQueue::Record.connection_pool.with_connection do + SolidQueue::Record.transaction do + NonOverlappingJob.set(acquire: acquire).perform_later("shared") + enqueued.set + + done.wait(1.second) + done.set? + end + end + end + + assert enqueued.wait(5.seconds), "The job inside the transaction was never enqueued" + block.call + done.set + + holder.value + end + + # A job holding the semaphore for the shared key, as one being performed does. + def enqueue_running_job + NonOverlappingJob.perform_later("shared") + SolidQueue::Job.last.tap { |job| assert job.ready? } + end + + # Two transactions, each enqueueing a job for one key and then, once the + # other has done the same, a job for the other's key. Answers with the + # errors raised in either of them. + def enqueue_in_opposite_orders(acquire:) + took_first = [ Concurrent::Event.new, Concurrent::Event.new ] + + threads = [ %w[ one two ], %w[ two one ] ].each_with_index.map do |(first, second), index| + Thread.new do + SolidQueue::Record.connection_pool.with_connection do + SolidQueue::Record.transaction do + NonOverlappingJob.set(acquire: acquire).perform_later(first) + took_first[index].set + took_first[1 - index].wait(5.seconds) + + NonOverlappingJob.set(acquire: acquire).perform_later(second) + end + end + + nil + rescue => error + error + end + end + + threads.map(&:value).compact + end +end diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb index 7e9c1568f..341d358c8 100644 --- a/test/models/solid_queue/job_test.rb +++ b/test/models/solid_queue/job_test.rb @@ -34,6 +34,22 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob limits_concurrency key: ->(job_result, **) { job_result }, group: "DiscardingGroup", on_conflict: :discard end + class LockOnDispatchJob < NonOverlappingJob + limits_concurrency key: ->(job_result, **) { job_result }, acquire: :on_dispatch + end + + class DiscardableLockOnDispatchJob < NonOverlappingJob + limits_concurrency key: ->(job_result, **) { job_result }, on_conflict: :discard, acquire: :on_dispatch + end + + class LockOutsideTransactionJob < NonOverlappingJob + # Enqueued inside the transaction, which is what the choice is about. + # Rails 7.2 takes :never rather than false. + self.enqueue_after_transaction_commit = Rails.version.start_with?("7.2") ? :never : false if respond_to?(:enqueue_after_transaction_commit=) + + limits_concurrency key: ->(job_result, **) { job_result }, acquire: :outside_transaction + end + setup do @result = JobResult.create!(queue_name: "default") @discarded_concurrent_error = SolidQueue::Job::EnqueueError.new( @@ -375,7 +391,146 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob end end + test "a job acquiring its concurrency lock on dispatch is enqueued as scheduled, without touching the semaphore" do + assert_scheduled do + LockOnDispatchJob.perform_later(@result, name: "A") + end + + job = SolidQueue::Job.last + assert job.scheduled_at <= Time.current + assert_nil SolidQueue::Semaphore.find_by(key: job.concurrency_key) + end + + test "the dispatcher acquires the concurrency lock of jobs that defer it" do + assert_job_counts(scheduled: 2) do + 2.times { |i| LockOnDispatchJob.perform_later(@result, name: i.to_s) } + end + + SolidQueue::ScheduledExecution.dispatch_next_batch(10) + + first, second = SolidQueue::Job.last(2) + assert first.ready? + assert second.blocked? + assert_equal 0, SolidQueue::Semaphore.find_by(key: first.concurrency_key).value + end + + test "a job that discards on conflict and defers its lock is discarded when dispatched" do + assert_job_counts(scheduled: 2) do + 2.times { |i| DiscardableLockOnDispatchJob.perform_later(@result, name: i.to_s) } + end + + assert_difference -> { SolidQueue::Job.count }, -1 do + SolidQueue::ScheduledExecution.dispatch_next_batch(10) + end + + assert_equal 1, SolidQueue::ReadyExecution.count + assert_equal 0, SolidQueue::BlockedExecution.count + end + + test "an enqueue's choice of when to acquire the concurrency lock overrides the job's" do + assert_ready do + LockOnDispatchJob.set(acquire: :on_enqueue).perform_later(@result, name: "A") + end + + assert_scheduled do + NonOverlappingJob.set(acquire: :on_dispatch).perform_later(@result, name: "B") + end + end + + test "the global choice of when to acquire the concurrency lock applies to jobs that don't make their own" do + with_concurrency_lock_acquisition(:on_dispatch) do + assert_scheduled do + NonOverlappingJob.perform_later(@result, name: "A") + end + + assert_ready do + LockOnDispatchJob.set(acquire: :on_enqueue).perform_later(@result, name: "B") + end + end + end + + test "jobs without concurrency controls are dispatched right away whatever the choice" do + with_concurrency_lock_acquisition(:on_dispatch) do + assert_ready do + AddToBufferJob.set(acquire: :on_dispatch).perform_later(1) + end + end + end + + test "jobs without concurrency controls don't check for an open transaction" do + SolidQueue::Job.expects(:inside_transaction?).never + + with_concurrency_lock_acquisition(:outside_transaction) do + assert_ready do + AddToBufferJob.perform_later(1) + end + end + end + + test "a job acquiring its concurrency lock outside transactions defers it only inside one on Solid Queue's connection" do + assert_ready do + LockOutsideTransactionJob.perform_later(@result, name: "A") + end + + assert_scheduled do + SolidQueue::Record.transaction do + LockOutsideTransactionJob.perform_later(@result, name: "B") + end + end + + # The app's database is a different one here, so its transaction doesn't + # hold Solid Queue's locks: the job acquires its lock right away, and blocks + # behind A. + assert_blocked do + JobResult.transaction do + LockOutsideTransactionJob.perform_later(@result, name: "C") + end + end + end + + test "enqueue jobs in bulk deferring the concurrency lock for some of them" do + active_jobs = [ + NonOverlappingJob.new(@result), + NonOverlappingJob.new(@result).set(acquire: :on_dispatch), + LockOnDispatchJob.new(@result), + AddToBufferJob.new(1).set(acquire: :on_dispatch) + ] + + assert_job_counts(ready: 2, scheduled: 2) do + ActiveJob.perform_all_later(active_jobs) + end + + assert active_jobs.all?(&:successfully_enqueued?) + assert_equal active_jobs.values_at(1, 2).map(&:provider_job_id).sort, SolidQueue::ScheduledExecution.pluck(:job_id).sort + end + + test "an unknown choice of when to acquire the concurrency lock is refused" do + assert_raises(ArgumentError) do + Class.new(ApplicationJob) { limits_concurrency key: "key", acquire: :later } + end + + assert_raises(ArgumentError) do + NonOverlappingJob.set(acquire: :later).perform_later(@result) + end + + assert_raises(ArgumentError) do + SolidQueue.concurrency_lock_acquisition = :later + end + + assert_raises(ArgumentError) do + SolidQueue.concurrency_lock_acquisition = 1 + end + assert_equal :on_enqueue, SolidQueue.concurrency_lock_acquisition + end + private + def with_concurrency_lock_acquisition(value) + previous, SolidQueue.concurrency_lock_acquisition = SolidQueue.concurrency_lock_acquisition, value + yield + ensure + SolidQueue.concurrency_lock_acquisition = previous + end + def assert_ready(&block) assert_job_counts(ready: 1, &block) assert SolidQueue::Job.last.ready?