Merge pull request rails#51426 from Shopify/transaction-callbacks-2

Automatically delay Active Job enqueues to after commit

Author: byroot

.rubocop.yml

@@ -55,10 +55,6 @@ Rails/IndexWith:
 Style/AndOr:
   Enabled: true
 
-# Align `when` with `case`.
-Layout/CaseIndentation:
-  Enabled: true
-
 Layout/ClosingHeredocIndentation:
   Enabled: true
 

activejob/CHANGELOG.md

@@ -1,3 +1,30 @@
+*   Make Active Job transaction aware when used conjointly with Active Record.
+
+    A common mistake with Active Job is to enqueue jobs from inside a transaction,
+    causing them to potentially be picked and ran by another process, before the
+    transaction is committed, which result in various errors.
+
+    ```ruby
+    Topic.transaction do
+      topic = Topic.create(...)
+      NewTopicNotificationJob.perform_later(topic)
+    end
+    ```
+
+    Now Active Job will automatically defer the enqueuing to after the transaction is committed,
+    and drop the job if the transaction is rolled back.
+
+    Various queue implementations can chose to disable this behavior, and users can disable it,
+    or force it on a per job basis:
+
+    ```ruby
+    class NewTopicNotificationJob < ApplicationJob
+      self.enqueue_after_transaction_commit = false # or `true`
+    end
+    ```
+
+    *Jean Boussier*, *Cristian Bica*
+
 *   Do not trigger immediate loading of `ActiveJob::Base` when loading `ActiveJob::TestHelper`.
 
     *Maxime Réty*

activejob/lib/active_job.rb

@@ -39,6 +39,7 @@ module ActiveJob
   autoload :Arguments
   autoload :DeserializationError, "active_job/arguments"
   autoload :SerializationError, "active_job/arguments"
+  autoload :EnqueueAfterTransactionCommit
 
   eager_autoload do
     autoload :Serializers

activejob/lib/active_job/enqueue_after_transaction_commit.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module ActiveJob
+  module EnqueueAfterTransactionCommit # :nodoc:
+    extend ActiveSupport::Concern
+
+    included do
+      ##
+      # :singleton-method:
+      #
+      # Defines if enqueueing this job from inside an Active Record transaction
+      # automatically defers the enqueue to after the transaction commit.
+      #
+      # It can be set on a per job basis:
+      #  - `:always` forces the job to be deferred.
+      #  - `:never` forces the job to be queueed immediately
+      #  - `:default` let the queue adapter define the behavior (recommended).
+      class_attribute :enqueue_after_transaction_commit, instance_accessor: false, instance_predicate: false, default: :never
+
+      around_enqueue do |job, block|
+        after_transaction = case job.class.enqueue_after_transaction_commit
+        when :always
+          true
+        when :never
+          false
+        else # :default
+          queue_adapter.enqueue_after_transaction_commit?
+        end
+
+        if after_transaction
+          ActiveRecord.after_all_transactions_commit(&block)
+        else
+          block.call
+        end
+      end
+    end
+  end
+end

activejob/lib/active_job/enqueuing.rb

@@ -53,6 +53,15 @@ module ClassMethods
       # Job#arguments or false if the enqueue did not succeed.
       #
       # After the attempted enqueue, the job will be yielded to an optional block.
+      #
+      # If Active Job is used conjointly with Active Record, and #perform_later is called
+      # inside an Active Record transaction, then the enqueue is implictly defered to after
+      # the transaction is committed, or droped if it's rolled back. This behavior can
+      # be changed on a per job basis:
+      #
+      #  class NotificationJob < ApplicationJob
+      #    self.enqueue_after_transaction_commit = false
+      #  end
       def perform_later(...)
         job = job_or_instantiate(...)
         enqueue_result = job.enqueue

activejob/lib/active_job/queue_adapters.rb

@@ -114,6 +114,7 @@ module ActiveJob
   module QueueAdapters
     extend ActiveSupport::Autoload
 
+    autoload :AbstractAdapter
     autoload :AsyncAdapter
     autoload :InlineAdapter
     autoload :BackburnerAdapter

activejob/lib/active_job/queue_adapters/abstract_adapter.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module ActiveJob
+  module QueueAdapters
+    # = Active Job Abstract Adapter
+    #
+    # Active Job supports multiple job queue systems. ActiveJob::QueueAdapters::AbstractAdapter
+    # form the abstraction layer which makes this possible.
+    class AbstractAdapter
+      # Define whether enqueuing should implictly to after commit when called from
+      # inside a transaction. Most adapters should return true, but some adapters
+      # that use the same database as Active Record and are transaction aware can return
+      # false to continue enqueuing jobs are part of the transaction.
+      def enqueue_after_transaction_commit?
+        true
+      end
+
+      def enqueue(job)
+        raise NotImplementedError
+      end
+
+      def enqueue_at(job, timestamp)
+        raise NotImplementedError
+      end
+    end
+  end
+end

activejob/lib/active_job/queue_adapters/async_adapter.rb

@@ -30,7 +30,7 @@ module QueueAdapters
     # The adapter uses a {Concurrent Ruby}[https://github.com/ruby-concurrency/concurrent-ruby] thread pool to schedule and execute
     # jobs. Since jobs share a single thread pool, long-running jobs will block
     # short-lived jobs. Fine for dev/test; bad for production.
-    class AsyncAdapter
+    class AsyncAdapter < AbstractAdapter
       # See {Concurrent::ThreadPoolExecutor}[https://ruby-concurrency.github.io/concurrent-ruby/master/Concurrent/ThreadPoolExecutor.html] for executor options.
       def initialize(**executor_options)
         @scheduler = Scheduler.new(**executor_options)

activejob/lib/active_job/queue_adapters/backburner_adapter.rb

@@ -14,7 +14,7 @@ module QueueAdapters
     # To use Backburner set the queue_adapter config to +:backburner+.
     #
     #   Rails.application.config.active_job.queue_adapter = :backburner
-    class BackburnerAdapter
+    class BackburnerAdapter < AbstractAdapter
       def enqueue(job) # :nodoc:
         response = Backburner::Worker.enqueue(JobWrapper, [job.serialize], queue: job.queue_name, pri: job.priority)
         job.provider_job_id = response[:id] if response.is_a?(Hash)

activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb

@@ -15,7 +15,15 @@ module QueueAdapters
     # To use Delayed Job, set the queue_adapter config to +:delayed_job+.
     #
     #   Rails.application.config.active_job.queue_adapter = :delayed_job
-    class DelayedJobAdapter
+    class DelayedJobAdapter < AbstractAdapter
+      def initialize(enqueue_after_transaction_commit: false)
+        @enqueue_after_transaction_commit = enqueue_after_transaction_commit
+      end
+
+      def enqueue_after_transaction_commit? # :nodoc:
+        @enqueue_after_transaction_commit
+      end
+
       def enqueue(job) # :nodoc:
         delayed_job = Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name, priority: job.priority)
         job.provider_job_id = delayed_job.id