Dispatch blocked jobs after job perform finishes outside that transaction

Otherwise, if a job finishes just fine or fails, and something goes wrong when
dispatching blocked jobs, we'll fail to mark the job as finished or failed just
because we couldn't dispatch blocked jobs, but not because anything went wrong with the
job itself.

Also: test sequential jobs when some of them fail. We must continue processing jobs
normally.

Author: rosa

app/models/solid_queue/claimed_execution.rb

@@ -33,6 +33,8 @@ def perform
     else
       failed_with(result.error)
     end
+  ensure
+    job.dispatch_blocked_jobs
   end
 
   def release

app/models/solid_queue/job/concurrency_controls.rb

@@ -7,6 +7,12 @@ module ConcurrencyControls
         has_one :blocked_execution, dependent: :destroy
       end
 
+      def dispatch_blocked_jobs
+        if release_concurrency_lock
+          release_next_blocked_job
+        end
+      end
+
       private
         def acquire_concurrency_lock
           return true unless concurrency_limited?

app/models/solid_queue/job/executable.rb

@@ -28,7 +28,6 @@ def prepare_for_execution
 
       def finished
         touch(:finished_at)
-        dispatch_blocked_jobs
       end
 
       def finished?
@@ -37,7 +36,6 @@ def finished?
 
       def failed_with(exception)
         FailedExecution.create_or_find_by!(job_id: id, exception: exception)
-        dispatch_blocked_jobs
       end
 
       def discard
@@ -60,12 +58,6 @@ def dispatch
           end
         end
 
-        def dispatch_blocked_jobs
-          if release_concurrency_lock
-            release_next_blocked_job
-          end
-        end
-
         def schedule
           ScheduledExecution.create_or_find_by!(job_id: id)
         end

test/dummy/app/jobs/update_result_job.rb

@@ -1,11 +1,11 @@
 class UpdateResultJob < ApplicationJob
   include ActiveJob::ConcurrencyControls
 
-  def perform(job_result, name:, pause: nil)
+  def perform(job_result, name:, pause: nil, exception: nil)
     job_result.status += "s#{name}"
-    job_result.save!
 
     sleep(pause) if pause
+    raise exception.new if exception
 
     job_result.status += "c#{name}"
     job_result.save!

test/integration/concurrency_controls_test.rb

@@ -64,6 +64,22 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     assert_stored_sequence(@result, [ "C" ])
   end
 
+  test "run several jobs over the same record sequentially, with some of them failing" do
+    ("A".."F").each_with_index do |name, i|
+      # A, C, E will fail, for i= 0, 2, 4
+      SequentialUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds, exception: (RuntimeError if i.even?))
+    end
+
+    ("G".."K").each do |name|
+      SequentialUpdateResultJob.perform_later(@result, name: name)
+    end
+
+    wait_for_jobs_to_finish_for(4.seconds)
+    assert_equal 3, SolidQueue::FailedExecution.count
+
+    assert_stored_sequence @result, [ "B", "D", "F" ] + ("G".."K").to_a
+  end
+
   private
     def assert_stored_sequence(result, sequence)
       expected = "seq: " + sequence.map { |name| "s#{name}c#{name}"}.join