Release any orphaned claimed executions as part of the process prune

Orphaned executions shouldn't happen as part of regular operation but
some people have reported them. The only way I can think of this
happening could be that a worker gets deregistered while still working
normally, so that it happens to be claiming executions while the process
record is deleted. This means that the callback to release claimed
executions might not see yet the executions that are being claimed at
the same time, and as such, it won't release any. Or, in other words,
those executions would get claimed by a process that's getting deleted
and that once they're committed to the DB, it no longer exists.

To recover from this scenario, as well as checking for processes with an
expired heartbeat, the supervisor will also check for orphaned
executions and release them.

Author: rosa

app/models/solid_queue/claimed_execution.rb

@@ -3,6 +3,8 @@
 class SolidQueue::ClaimedExecution < SolidQueue::Execution
   belongs_to :process
 
+  scope :orphaned, -> { where.missing(:process) }
+
   class Result < Struct.new(:success, :error)
     def success?
       success

lib/solid_queue/supervisor.rb

@@ -22,7 +22,8 @@ def start
       run_callbacks(:boot) { boot }
 
       start_forks
-      launch_process_prune
+      launch_maintenance_task
+
       supervise
     rescue Processes::GracefulTerminationRequested
       graceful_termination
@@ -65,9 +66,12 @@ def start_forks
         configured_processes.each { |configured_process| start_fork(configured_process) }
       end
 
-      def launch_process_prune
-        @prune_task = Concurrent::TimerTask.new(run_now: true, execution_interval: SolidQueue.process_alive_threshold) { prune_dead_processes }
-        @prune_task.execute
+      def launch_maintenance_task
+        @maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: SolidQueue.process_alive_threshold) do
+          prune_dead_processes
+          release_orphaned_executions
+        end
+        @maintenance_task.execute
       end
 
       def shutdown
@@ -106,7 +110,7 @@ def quit_forks
       end
 
       def stop_process_prune
-        @prune_task&.shutdown
+        @maintenance_task&.shutdown
       end
 
       def delete_pidfile
@@ -117,6 +121,10 @@ def prune_dead_processes
         wrap_in_app_executor { SolidQueue::Process.prune }
       end
 
+      def release_orphaned_executions
+        wrap_in_app_executor { SolidQueue::ClaimedExecution.orphaned.release_all }
+      end
+
       def start_fork(configured_process)
         configured_process.supervised_by process
 

test/integration/processes_lifecycle_test.rb

@@ -20,8 +20,8 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
   end
 
   test "enqueue jobs in multiple queues" do
-    6.times.map { |i| enqueue_store_result_job("job_#{i}") }
-    6.times.map { |i| enqueue_store_result_job("job_#{i}", :default) }
+    6.times { |i| enqueue_store_result_job("job_#{i}") }
+    6.times { |i| enqueue_store_result_job("job_#{i}", :default) }
 
     wait_for_jobs_to_finish_for(0.5.seconds)
 

test/unit/supervisor_test.rb

@@ -67,7 +67,7 @@ class SupervisorTest < ActiveSupport::TestCase
     wait_for_process_termination_with_timeout(pid, exitstatus: 1)
   end
 
-  test "deletes previous pidfile if the owner is dead" do
+  test "delete previous pidfile if the owner is dead" do
     pid = run_supervisor_as_fork(mode: :all)
     wait_for_registered_processes(4)
 
@@ -87,6 +87,32 @@ class SupervisorTest < ActiveSupport::TestCase
     terminate_process(pid)
   end
 
+  test "release orphaned executions" do
+    3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
+    process = SolidQueue::Process.register(kind: "Worker", pid: 42)
+
+    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+
+    assert_equal 3, SolidQueue::ClaimedExecution.count
+    assert_equal 0, SolidQueue::ReadyExecution.count
+
+    assert_equal [ process.id ], SolidQueue::ClaimedExecution.last(3).pluck(:process_id).uniq
+
+    # Simnulate orphaned executions by just wiping the claiming process
+    process.delete
+
+    pid = run_supervisor_as_fork
+    wait_for_registered_processes(4)
+    assert_registered_supervisor(pid)
+
+    terminate_process(pid)
+
+    skip_active_record_query_cache do
+      assert_equal 0, SolidQueue::ClaimedExecution.count
+      assert_equal 3, SolidQueue::ReadyExecution.count
+    end
+  end
+
   private
     def assert_registered_workers(count, supervisor_pid:, **metadata)
       skip_active_record_query_cache do