Report batches that were never sealed instead of completing them

README.md

@@ -34,6 +34,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Batch jobs](#batch-jobs)
   - [Batch progress and counters](#batch-progress-and-counters)
   - [Batch maintenance](#batch-maintenance)
+    - [Stalled batches](#stalled-batches)
   - [Clearing batches](#clearing-batches)
   - [Upgrading existing installations](#upgrading-existing-installations)
 - [Puma plugin](#puma-plugin)
@@ -763,7 +764,7 @@ Batches track `total_jobs`, `completed_jobs`, `failed_jobs` and `pending_jobs`,
 
 ### Batch maintenance
 
-Batch completion is normally detected as jobs finish, without ever locking the batch row outside a single once-per-batch moment. A few edge cases can't trigger that detection: jobs removed via bulk discards (which delete jobs without callbacks), a process that crashed after enqueueing jobs but before starting its batch, or a completion whose callback enqueueing failed and rolled back.
+Batch completion is normally detected as jobs finish, without ever locking the batch row outside a single once-per-batch moment. A couple of edge cases can't trigger that detection: jobs removed via bulk discards (which delete jobs without callbacks), or a completion whose callback enqueueing failed and rolled back.
 
 The dispatcher sweeps these up automatically via `SolidQueue::Batch.sweep_stalled`, as part of its regular maintenance (every `concurrency_maintenance_interval` seconds, sharing a single maintenance timer and database connection). If you disable `batch_maintenance` (or don't run a dispatcher), you can run the sweep yourself, for example as a [recurring task](#recurring-tasks):
 
@@ -773,6 +774,24 @@ batch_maintenance:
   schedule: every 5 minutes
 ```
 
+#### Stalled batches
+
+Maintenance only completes batches their creator *sealed*—batches `SolidQueue::Batch.enqueue` finished filling. A batch created outside any transaction is sealed in the same write as its row, so a process dying mid-creation can't leave an unsealed batch behind. One created inside a transaction can: sealing waits for the commit, and a transaction that's still open, rolled back, or died uncommitted leaves the batch unsealed—indistinguishably so.
+
+Completing an unsealed batch would lose work: it finishes with whatever happened to have landed, fires its callbacks, and enqueues still pending on the transaction then raise `SolidQueue::Batch::AlreadyFinished`. Rather than guess whether more jobs are coming, the sweep emits a `stalled_batch.solid_queue` event for each one and waits.
+
+An unsealed batch doesn't wait forever. One still unsealed a day after creation came from a transaction that will never commit, so it ends the way its data did: the sweep removes it, as if it was never created. No callbacks fire, and nothing accumulates or needs monitoring. Jobs that reached the queue before the rollback have already run and stay untouched—the same behaviour as jobs enqueued outside a batch in a rolled-back transaction.
+
+In between, batches are yours to inspect or resolve early:
+
+```ruby
+SolidQueue::Batch.stalled                          # unsealed for more than 5 minutes
+SolidQueue::Batch.stalled.find_each(&:start)       # complete them yourself
+SolidQueue::Batch.stalled.each(&:destroy)          # or discard them early
+```
+
+Jobs already in a stalled batch run normally—only the batch's own completion waits. Keep transactions that create batches shorter than the expiry window (`expire_after` on `sweep_stalled`, one day by default): a transaction that outlives it and then commits finds its batch gone and raises `SolidQueue::Batch::AlreadyFinished` from the deferred enqueues.
+
 ### Clearing batches
 
 Finished, non-failed batches are cleared with `SolidQueue::Batch.clear_finished_in_batches` after `config.solid_queue.clear_finished_jobs_after`, but only when you invoke it. Failed batches are kept, like failed jobs, so you can inspect them. Installing Solid Queue configures [a recurring task](#recurring-tasks) that clears finished jobs every hour; you can add a matching entry for batches to your `recurring.yml`:

app/models/solid_queue/batch.rb

@@ -63,11 +63,22 @@ def enqueue(&block)
         raise AlreadyFinished, "Can't enqueue an already finished batch"
       end
 
+      # Decided before opening our own transaction: with none open anywhere,
+      # nothing outside this method can roll the batch back or defer its jobs.
+      seal_on_create = no_enclosing_transactions?
+
       transaction do
         save! if new_record?
 
         self.class.wrap_in_batch_context(id) { block&.call(self) }
 
+        # Sealing in the same transaction as the batch row makes the crashed
+        # creator unrepresentable: either nothing commits, or a sealed batch
+        # and its jobs commit together. Only batches that got jobs qualify;
+        # one with none may still be waiting on deferred enqueues, and sealed
+        # empty means complete.
+        seal_if_filled if seal_on_create
+
         if ActiveRecord.respond_to?(:after_all_transactions_commit)
           ActiveRecord.after_all_transactions_commit { start }
         end
@@ -106,6 +117,14 @@ def mark_as_enqueued
         Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current)
       end
 
+      def seal_if_filled
+        Batch.where(id: id, enqueued_at: nil).where.not(total_jobs: 0).update_all(enqueued_at: Time.current)
+      end
+
+      def no_enclosing_transactions?
+        ActiveRecord.respond_to?(:all_open_transactions) && ActiveRecord.all_open_transactions.none?
+      end
+
       def finalize
         reload
 

app/models/solid_queue/batch/sweepable.rb

@@ -2,22 +2,38 @@
 
 module SolidQueue
   class Batch
-    # Repairs batches that the regular completion detection can't finish on
-    # its own: jobs removed via bulk discards, processes that crashed after
-    # enqueueing jobs but before starting their batch, or completions whose
-    # callback enqueueing failed and rolled back.
+    # Repairs batches that the regular completion detection can't finish on its
+    # own: jobs removed via bulk discards, or completions whose callback
+    # enqueueing failed and rolled back.
+    #
+    # Repair here never invents state. A batch is only completed once its creator
+    # sealed it by calling #start, because "sealed" is the only thing that makes
+    # an empty batch meaningfully complete rather than merely unfilled. Batches
+    # that were never sealed are reported, not finished—see #report_stalled_batches.
     module Sweepable
       extend ActiveSupport::Concern
 
+      included do
+        scope :unsealed, -> { unfinished.where(enqueued_at: nil) }
+      end
+
       class_methods do
-        def sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
-          SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, stale_executions: 0, finished_batches: 0, started_batches: 0) do |payload|
+        def sweep_stalled(stalled_for: 5.minutes, expire_after: 1.day, batch_size: 500)
+          SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, expire_after: expire_after, stale_executions: 0, finished_batches: 0, expired_batches: 0, stalled_batches: 0) do |payload|
             payload[:stale_executions] = sweep_stale_executions(batch_size:)
             payload[:finished_batches] = finish_stalled_batches(batch_size:)
-            payload[:started_batches] = start_stalled_batches(stalled_for:, batch_size:)
+            payload[:expired_batches] = expire_abandoned_batches(expire_after:, batch_size:)
+            payload[:stalled_batches] = report_stalled_batches(stalled_for:, batch_size:)
           end
         end
 
+        # Batches their creator never sealed, and hasn't sealed for a while. Use
+        # this to find them, and SolidQueue::Batch#start to adopt one deliberately
+        # once you've established its creator is gone for good.
+        def stalled(stalled_for: 5.minutes)
+          unsealed.where(created_at: ...stalled_for.ago)
+        end
+
         private
           # BatchExecution rows represent outstanding work. A row for a resolved
           # job violates that invariant, so remove it immediately; destroy's
@@ -35,7 +51,9 @@ def sweep_stale_executions(batch_size:)
             swept
           end
 
-          # A started batch with no tracking rows left can finish
+          # A sealed batch with no tracking rows left can finish. Sealed is the
+          # load-bearing word: its creator got far enough to declare the batch
+          # complete, so an empty one really is done.
           def finish_stalled_batches(batch_size:)
             finished = 0
 
@@ -47,16 +65,48 @@ def finish_stalled_batches(batch_size:)
             finished
           end
 
-          # A batch that crashed between creation and start never got enqueued
-          def start_stalled_batches(stalled_for:, batch_size:)
-            started = 0
+          # A batch unsealed long past any reasonable transaction was created in
+          # a transaction that will never commit: its data rolled back, so the
+          # batch ends the same way—removed, as if never created. Not completing
+          # it means no callbacks ever fire over rolled-back work; not keeping
+          # it means nothing accumulates or needs monitoring. Jobs that reached
+          # the queue before the rollback already ran and stay untouched, just
+          # like jobs enqueued outside a batch in a rolled-back transaction.
+          #
+          # A transaction that outlives expire_after and then commits raises
+          # AlreadyFinished from its deferred enqueues: loud, and without
+          # firing success callbacks over lost work.
+          def expire_abandoned_batches(expire_after:, batch_size:)
+            expired = 0
+
+            unsealed.where(created_at: ...expire_after.ago).find_each(batch_size: batch_size) do |batch|
+              expired += 1
+              batch.destroy
+            end
+
+            expired
+          end
+
+          # Batches created outside any transaction seal in the same write as
+          # their row, so a crashed creator can't leave one behind. An unsealed
+          # batch therefore came from inside a transaction—one still filling it,
+          # rolled back, or died uncommitted—or is still waiting on deferred
+          # enqueues. Completing any of those loses work: the batch finishes
+          # with whatever happened to have landed, fires its callbacks, and the
+          # real enqueues then raise AlreadyFinished. Since "still coming" and
+          # "never coming" are indistinguishable here, report them and let an
+          # operator decide, rather than guessing and reporting success for
+          # work that never ran.
+          def report_stalled_batches(stalled_for:, batch_size:)
+            stalled_batches = stalled(stalled_for: stalled_for)
+            count = stalled_batches.count
+            return 0 if count.zero?
 
-            unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
-              started += 1
-              batch.start
+            stalled_batches.find_each(batch_size: batch_size) do |batch|
+              SolidQueue.instrument(:stalled_batch, batch_id: batch.id, created_at: batch.created_at, total_jobs: batch.total_jobs)
             end
 
-            started
+            count
           end
       end
     end

lib/solid_queue/log_subscriber.rb

@@ -47,7 +47,7 @@ def finish_batch(event)
   end
 
   def sweep_stalled_batches(event)
-    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :started_batches))
+    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :expired_batches, :stalled_batches))
   end
 
   def batch_progress_error(event)

test/integration/batch_lifecycle_test.rb

@@ -127,6 +127,92 @@ def perform
     assert_finished_in_order(job!(job1), batch1.reload)
   end
 
+  test "a batch filled from a transaction that outlives the stalled window still completes correctly" do
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
+
+    ApplicationJob.enqueue_after_transaction_commit = true
+
+    batch = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+
+      batch = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("late")) do
+        AddToBufferJob.perform_later("late")
+      end
+
+      # The batch row commits before this transaction does, so maintenance
+      # running elsewhere can reach it while the enqueues are still pending
+      SolidQueue::Batch.where(id: batch.id).update_all(created_at: 10.minutes.ago)
+      SolidQueue::Batch.sweep_stalled
+
+      assert_not batch.reload.finished?, "maintenance must not complete a batch still being filled"
+    end
+
+    assert_equal 1, batch.reload.total_jobs
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert batch.reload.finished?
+    assert_equal [ "late", "late: 1 jobs succeeded!" ].sort, JobBuffer.values.sort
+  end
+
+  test "a batch from a rolled-back transaction never reports success" do
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
+
+    batch = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+
+      batch = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("phantom")) do
+        AddToBufferJob.perform_later("rolled back")
+      end
+
+      raise ActiveRecord::Rollback
+    end
+
+    # Maintenance finds the leftover batch old and unstarted, and leaves it alone
+    SolidQueue::Batch.where(id: batch.id).update_all(created_at: 10.minutes.ago)
+    SolidQueue::Batch.sweep_stalled
+
+    @dispatcher.start
+    @worker.start
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_not batch.reload.finished?
+    assert_nil batch.enqueued_at
+    assert_not_includes JobBuffer.values, "phantom: 1 jobs succeeded!"
+  end
+
+  test "a batch whose creator dies right after enqueue is already complete" do
+    skip unless ActiveRecord.respond_to?(:all_open_transactions)
+
+    # The deferred start never runs, as if the process died the moment enqueue returned
+    SolidQueue::Batch.any_instance.stubs(:start)
+
+    jobs_in_transaction = nil
+    batch = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("sealed")) do
+      AddToBufferJob.perform_later("sealed")
+      jobs_in_transaction = SolidQueue::Job.count
+    end
+
+    skip "enqueues are deferred here, so sealing waits for start" if jobs_in_transaction.zero?
+
+    assert batch.reload.enqueued?, "the batch sealed in the same commit as its row"
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert batch.reload.finished?
+    assert_equal [ "sealed", "sealed: 1 jobs succeeded!" ].sort, JobBuffer.values.sort
+  end
+
   test "prebuilt jobs capture their batch before enqueue is deferred" do
     skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1