Enqueue all jobs in perform_all_later within the same transaction

This allows even further DRYing and simplification, as we can unify the
dispatching done when moving a scheduled job batch to ready, having to
deal with concurrency-limited jobs, and the same thing when enqueuing
multiple jobs.

Author: rosa

app/models/solid_queue/job.rb

@@ -12,34 +12,16 @@ class Job < Record
 
     class << self
       def enqueue_all(active_jobs)
-        scheduled_jobs, immediate_jobs = active_jobs.partition(&:scheduled_at)
-        with_concurrency_limits, without_concurrency_limits = immediate_jobs.partition(&:concurrency_limited?)
-
-        schedule_all_at_once(scheduled_jobs)
-        enqueue_all_at_once(without_concurrency_limits)
-        enqueue_one_by_one(with_concurrency_limits)
-      end
-
-      def schedule_all_at_once(active_jobs)
-        transaction do
-          inserted_jobs = create_all_from_active_jobs(active_jobs)
-          schedule_all(inserted_jobs)
-        end
-      end
-
-      def enqueue_all_at_once(active_jobs)
         transaction do
-          inserted_jobs = create_all_from_active_jobs(active_jobs)
-          dispatch_all_at_once(inserted_jobs)
+          jobs = create_all_from_active_jobs(active_jobs)
+          prepare_all_for_execution(jobs)
         end
       end
 
-      def enqueue_one_by_one(active_jobs)
-        active_jobs.each { |active_job| enqueue(active_job) }
-      end
-
       def enqueue(active_job, scheduled_at: Time.current)
-        create!(**attributes_from_active_job(active_job).reverse_merge(scheduled_at: scheduled_at)).tap do |job|
+        active_job.scheduled_at = scheduled_at
+
+        create_from_active_job(active_job).tap do |job|
           active_job.provider_job_id = job.id
         end
       end
@@ -48,6 +30,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))
+        end
+
         def create_all_from_active_jobs(active_jobs)
           job_rows = active_jobs.map { |job| attributes_from_active_job(job) }
           insert_all(job_rows)
@@ -56,18 +42,14 @@ def create_all_from_active_jobs(active_jobs)
 
         def attributes_from_active_job(active_job)
           {
-            queue_name: active_job.queue_name,
+            queue_name: active_job.queue_name || DEFAULT_QUEUE_NAME,
             active_job_id: active_job.job_id,
-            priority: active_job.priority,
+            priority: active_job.priority || DEFAULT_PRIORITY,
             scheduled_at: active_job.scheduled_at,
             class_name: active_job.class.name,
             arguments: active_job.serialize,
             concurrency_key: active_job.concurrency_key
-          }.compact.with_defaults(defaults)
-        end
-
-        def defaults
-          { queue_name: DEFAULT_QUEUE_NAME, priority: DEFAULT_PRIORITY }
+          }
         end
     end
   end

app/models/solid_queue/job/executable.rb

@@ -21,13 +21,32 @@ module Executable
       end
 
       class_methods do
-        def dispatch_all_at_once(jobs)
-          ReadyExecution.create_all_from_jobs(jobs)
+        def prepare_all_for_execution(jobs)
+          due, not_yet_due = jobs.partition(&:due?)
+
+          dispatch_all(due)
+          schedule_all(not_yet_due)
+        end
+
+        def dispatch_all(jobs)
+          with_concurrency_limits, without_concurrency_limits = jobs.partition(&:concurrency_limited?)
+
+          dispatch_all_at_once(without_concurrency_limits)
+          dispatch_all_one_by_one(with_concurrency_limits)
         end
 
         def schedule_all(jobs)
           ScheduledExecution.create_all_from_jobs(jobs)
         end
+
+        private
+          def dispatch_all_at_once(jobs)
+            ReadyExecution.create_all_from_jobs jobs
+          end
+
+          def dispatch_all_one_by_one(jobs)
+            jobs.each(&:dispatch)
+          end
       end
 
       %w[ ready claimed failed scheduled ].each do |status|
@@ -60,6 +79,10 @@ def finished?
         finished_at.present?
       end
 
+      def due?
+        scheduled_at.nil? || scheduled_at <= Time.current
+      end
+
       def discard
         destroy unless claimed?
       end
@@ -73,10 +96,6 @@ def failed_with(exception)
       end
 
       private
-        def due?
-          scheduled_at.nil? || scheduled_at <= Time.current
-        end
-
         def schedule
           ScheduledExecution.create_or_find_by!(job_id: id)
         end

app/models/solid_queue/scheduled_execution.rb

@@ -22,25 +22,14 @@ def dispatch_next_batch(batch_size)
       private
         def dispatch_batch(job_ids)
           jobs = Job.where(id: job_ids)
-          with_concurrency_limits, without_concurrency_limits = jobs.partition(&:concurrency_limited?)
-
-          dispatch_at_once(without_concurrency_limits)
-          dispatch_one_by_one(with_concurrency_limits)
+          Job.dispatch_all(jobs)
 
           successfully_dispatched(job_ids).tap do |dispatched_job_ids|
             where(job_id: dispatched_job_ids).delete_all
             SolidQueue.logger.info("[SolidQueue] Dispatched scheduled batch with #{dispatched_job_ids.size} jobs")
           end
         end
 
-        def dispatch_at_once(jobs)
-          ReadyExecution.create_all_from_jobs jobs
-        end
-
-        def dispatch_one_by_one(jobs)
-          jobs.each(&:dispatch)
-        end
-
         def successfully_dispatched(job_ids)
           dispatched_and_ready(job_ids) + dispatched_and_blocked(job_ids)
         end