Remove support for nested batches for now

* We still track it, but it was causing alot of race conditions while trying to keep exclusively in callbacks. Running in a job or worker/dispatcher it works easily, but adds more overhead to the code and processing

* Move to explicit timestamp fields instead of status fields so it's easier to track specifics of batch transitions

* Move batches lower in the schema, after current models

Author: jpcamara

CLAUDE.md

@@ -2,9 +2,7 @@
 
 module SolidQueue
   class Batch
-    class CleanupJob < ApplicationJob
-      queue_as :background
-
+    class CleanupJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base)
       discard_on ActiveRecord::RecordNotFound
 
       def perform(job_batch)

app/jobs/solid_queue/batch/cleanup_job.rb

@@ -2,11 +2,9 @@
 
 module SolidQueue
   class Batch < Record
-    STATUSES = %w[pending processing completed failed]
-
     include Trackable
 
-    belongs_to :parent_batch, foreign_key: :parent_batch_id, class_name: "SolidQueue::Batch", optional: true
+    belongs_to :parent_batch, foreign_key: :parent_batch_id, primary_key: :batch_id, class_name: "SolidQueue::Batch", optional: true
     has_many :jobs, foreign_key: :batch_id, primary_key: :batch_id
     has_many :batch_executions, foreign_key: :batch_id, primary_key: :batch_id, class_name: "SolidQueue::BatchExecution"
     has_many :child_batches, foreign_key: :parent_batch_id, primary_key: :batch_id, class_name: "SolidQueue::Batch"
@@ -16,10 +14,9 @@ class Batch < Record
     serialize :on_failure, coder: JSON
     serialize :metadata, coder: JSON
 
-    validates :status, inclusion: { in: STATUSES }
-
     after_initialize :set_batch_id
     before_create :set_parent_batch_id
+    after_commit :start_batch, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }
 
     mattr_accessor :maintenance_queue_name
     self.maintenance_queue_name = "default"
@@ -33,12 +30,10 @@ def enqueue(&block)
         block.call(self)
       end
 
-      if ActiveSupport.respond_to?(:after_all_transactions_commit)
+      if ActiveRecord.respond_to?(:after_all_transactions_commit)
         ActiveRecord.after_all_transactions_commit do
-          start_monitoring
+          start_batch
         end
-      else
-        start_monitoring
       end
     end
 
@@ -55,22 +50,16 @@ def on_finish=(value)
     end
 
     def check_completion!
-      return if finished?
+      return if finished? || !ready?
 
       with_lock do
-        return if finished_at?
+        return if finished_at? || !ready?
 
         if pending_jobs == 0
-          unfinished_children = child_batches.where.not(status: %w[completed failed]).count
-
-          if unfinished_children == 0
-            new_status = failed_jobs > 0 ? "failed" : "completed"
-            update!(status: new_status, finished_at: Time.current)
-            execute_callbacks
-          end
-        elsif status == "pending" && (completed_jobs > 0 || failed_jobs > 0)
-          # Move from pending to processing once any job completes
-          update!(status: "processing")
+          finished_attributes = { finished_at: Time.current }
+          finished_attributes[:failed_at] = Time.current if failed_jobs > 0
+          update!(finished_attributes)
+          execute_callbacks
         end
       end
     end
@@ -108,28 +97,19 @@ def perform_completion_job(job_field, attrs)
       end
 
       def execute_callbacks
-        if status == "failed"
+        if failed_at?
           perform_completion_job(:on_failure, {}) if on_failure.present?
-        elsif status == "completed"
+        else
           perform_completion_job(:on_success, {}) if on_success.present?
         end
 
         perform_completion_job(:on_finish, {}) if on_finish.present?
 
         clear_unpreserved_jobs
-
-        check_parent_completion!
       end
 
       def clear_unpreserved_jobs
-        SolidQueue::Batch::CleanupJob.perform_later(self) unless SolidQueue.preserve_finished_jobs?
-      end
-
-      def check_parent_completion!
-        if parent_batch_id.present?
-          parent = Batch.find_by(batch_id: parent_batch_id)
-          parent&.check_completion! unless parent&.finished?
-        end
+        SolidQueue::Batch::CleanupJob.set(queue: self.class.maintenance_queue_name || "default").perform_later(self) unless SolidQueue.preserve_finished_jobs?
       end
 
       def enqueue_empty_job
@@ -138,15 +118,9 @@ def enqueue_empty_job
         end
       end
 
-      def enqueue_monitor_job
-        Batch.wrap_in_batch_context(nil) do
-          BatchMonitorJob.set(queue: self.class.maintenance_queue_name || "default").perform_later(batch_id: batch_id)
-        end
-      end
-
-      def start_monitoring
+      def start_batch
         enqueue_empty_job if reload.total_jobs == 0
-        enqueue_monitor_job
+        update!(enqueued_at: Time.current)
       end
 
     class << self

app/jobs/solid_queue/batch_monitor_job.rb

@@ -6,24 +6,26 @@ module Trackable
       extend ActiveSupport::Concern
 
       included do
-        scope :pending, -> { where(status: "pending") }
-        scope :processing, -> { where(status: "processing") }
-        scope :completed, -> { where(status: "completed") }
-        scope :failed, -> { where(status: "failed") }
-        scope :finished, -> { where(status: %w[completed failed]) }
-        scope :unfinished, -> { where(status: %w[pending processing]) }
+        scope :finished, -> { where.not(finished_at: nil) }
+        scope :succeeded, -> { finished.where(failed_at: nil) }
+        scope :unfinished, -> { where(finished_at: nil) }
+        scope :failed, -> { where.not(failed_at: nil) }
       end
 
-      def finished?
-        status.in?(%w[completed failed])
+      def failed?
+        failed_at.present?
+      end
+
+      def succeeded?
+        finished? && !failed?
       end
 
-      def processing?
-        status == "processing"
+      def finished?
+        finished_at.present?
       end
 
-      def pending?
-        status == "pending"
+      def ready?
+        enqueued_at.present?
       end
 
       def progress_percentage

app/models/solid_queue/batch.rb

@@ -42,6 +42,9 @@ def process_job_completion(job, status)
             )
           end
         end
+
+        batch = Batch.find_by(batch_id: batch_id)
+        batch.check_completion! if batch.present?
       end
 
       private

app/models/solid_queue/batch/trackable.rb

@@ -26,32 +26,6 @@
     t.index [ "job_id" ], name: "index_solid_queue_failed_executions_on_job_id", unique: true
   end
 
-  create_table "solid_queue_batches", force: :cascade do |t|
-    t.string "batch_id", null: false
-    t.string "parent_batch_id"
-    t.text "on_finish"
-    t.text "on_success"
-    t.text "on_failure"
-    t.text "metadata"
-    t.integer "total_jobs", default: 0, null: false
-    t.integer "pending_jobs", default: 0, null: false
-    t.integer "completed_jobs", default: 0, null: false
-    t.integer "failed_jobs", default: 0, null: false
-    t.string "status", default: "pending", null: false
-    t.datetime "finished_at"
-    t.datetime "created_at", null: false
-    t.datetime "updated_at", null: false
-    t.index [ "batch_id" ], name: "index_solid_queue_batches_on_batch_id", unique: true
-    t.index [ "parent_batch_id" ], name: "index_solid_queue_batches_on_parent_batch_id"
-  end
-
-  create_table "solid_queue_batch_executions", force: :cascade do |t|
-    t.bigint "job_id", null: false
-    t.string "batch_id", null: false
-    t.datetime "created_at", null: false
-    t.index [ "job_id" ], name: "index_solid_queue_batch_executions_on_job_id", unique: true
-  end
-
   create_table "solid_queue_jobs", force: :cascade do |t|
     t.string "queue_name", null: false
     t.string "class_name", null: false
@@ -148,6 +122,33 @@
     t.index [ "key" ], name: "index_solid_queue_semaphores_on_key", unique: true
   end
 
+  create_table "solid_queue_batches", force: :cascade do |t|
+    t.string "batch_id"
+    t.string "parent_batch_id"
+    t.text "on_finish"
+    t.text "on_success"
+    t.text "on_failure"
+    t.text "metadata"
+    t.integer "total_jobs", default: 0, null: false
+    t.integer "pending_jobs", default: 0, null: false
+    t.integer "completed_jobs", default: 0, null: false
+    t.integer "failed_jobs", default: 0, null: false
+    t.datetime "enqueued_at"
+    t.datetime "finished_at"
+    t.datetime "failed_at"
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["batch_id"], name: "index_solid_queue_batches_on_batch_id", unique: true
+    t.index ["parent_batch_id"], name: "index_solid_queue_batches_on_parent_batch_id"
+  end
+
+  create_table "solid_queue_batch_executions", force: :cascade do |t|
+    t.bigint "job_id", null: false
+    t.string "batch_id", null: false
+    t.datetime "created_at", null: false
+    t.index [ "job_id" ], name: "index_solid_queue_batch_executions_on_job_id", unique: true
+  end
+
   add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade

app/models/solid_queue/batch_execution.rb

@@ -38,32 +38,6 @@
     t.index ["job_id"], name: "index_solid_queue_failed_executions_on_job_id", unique: true
   end
 
-  create_table "solid_queue_batches", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
-    t.string "batch_id"
-    t.string "parent_batch_id"
-    t.text "on_finish"
-    t.text "on_success"
-    t.text "on_failure"
-    t.text "metadata"
-    t.integer "total_jobs", default: 0, null: false
-    t.integer "pending_jobs", default: 0, null: false
-    t.integer "completed_jobs", default: 0, null: false
-    t.integer "failed_jobs", default: 0, null: false
-    t.string "status", default: "pending", null: false
-    t.datetime "finished_at"
-    t.datetime "created_at", null: false
-    t.datetime "updated_at", null: false
-    t.index ["batch_id"], name: "index_solid_queue_batches_on_batch_id", unique: true
-    t.index ["parent_batch_id"], name: "index_solid_queue_batches_on_parent_batch_id"
-  end
-
-  create_table "solid_queue_batch_executions", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
-    t.bigint "job_id", null: false
-    t.string "batch_id", null: false
-    t.datetime "created_at", null: false
-    t.index ["job_id"], name: "index_solid_queue_batch_executions_on_job_id", unique: true
-  end
-
   create_table "solid_queue_jobs", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
     t.string "queue_name", null: false
     t.string "class_name", null: false
@@ -160,6 +134,33 @@
     t.index ["key"], name: "index_solid_queue_semaphores_on_key", unique: true
   end
 
+  create_table "solid_queue_batches", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.string "batch_id"
+    t.string "parent_batch_id"
+    t.text "on_finish"
+    t.text "on_success"
+    t.text "on_failure"
+    t.text "metadata"
+    t.integer "total_jobs", default: 0, null: false
+    t.integer "pending_jobs", default: 0, null: false
+    t.integer "completed_jobs", default: 0, null: false
+    t.integer "failed_jobs", default: 0, null: false
+    t.datetime "enqueued_at"
+    t.datetime "finished_at"
+    t.datetime "failed_at"
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["batch_id"], name: "index_solid_queue_batches_on_batch_id", unique: true
+    t.index ["parent_batch_id"], name: "index_solid_queue_batches_on_parent_batch_id"
+  end
+
+  create_table "solid_queue_batch_executions", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.bigint "job_id", null: false
+    t.string "batch_id", null: false
+    t.datetime "created_at", null: false
+    t.index ["job_id"], name: "index_solid_queue_batch_executions_on_job_id", unique: true
+  end
+
   add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade