Implement Active Job enqueue_after_transaction_commit

A fairly common mistake with Rails is to enqueue a job from inside a
transaction, with a record as argumemnt, which then lead to a RecordNotFound
error when picked up by the queue.

This is even one of the arguments advanced for job runners backed by the
database such as `solid_queue`, `delayed_job` or `good_job`.

But relying on this is undesirable in my opinion as it makes the Active Job
abstraction leaky, and if in the future you need to migrate to another backend
or even just move the queue to a separate database, you may experience a lot of
race conditions of the sort.

To resolve this problem globally, we can make Active Job optionally transaction
aware, and automatically defer job queueing to `after_commit`.

Co-Authored-By: Cristian Bica <[email protected]>

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