Let jobs acquire their concurrency lock when dispatched

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:

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,15 +57,39 @@ 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
           end
           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)

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

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)

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)

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