DRY a bit releasing claimed executions by both supervisors

Author: rosa

lib/solid_queue/async_supervisor.rb

@@ -3,43 +3,35 @@
 module SolidQueue
   class AsyncSupervisor < Supervisor
     private
-      def check_and_replace_terminated_processes
-        terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
-        terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
-      end
 
-      def replace_thread(thread_id, instance)
-        SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
-          payload[:thread] = instance
-          handle_claimed_jobs_by(terminated_instance, thread)
+    def check_and_replace_terminated_processes
+      terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
+      terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
+    end
 
-          start_process(configured_processes.delete(thread_id))
-        end
-      end
+    def replace_thread(thread_id, instance)
+      SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
+        payload[:thread] = instance
 
-      def perform_graceful_termination
-        process_instances.values.each(&:stop)
+        error = Processes::ThreadTerminatedError.new(terminated_instance.name)
+        release_claimed_jobs_by(terminated_instance, with_error: error)
 
-        Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+        start_process(configured_processes.delete(thread_id))
       end
+    end
 
-      def perform_immediate_termination
-        exit!
-      end
+    def perform_graceful_termination
+      process_instances.values.each(&:stop)
 
-      def all_processes_terminated?
-        process_instances.values.none?(&:alive?)
-      end
+      Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+    end
 
-      # When a supervised thread terminates unexpectedly, mark all executions
-      # it had claimed as failed so they can be retried by another worker.
-      def handle_claimed_jobs_by(terminated_instance, thread)
-        wrap_in_app_executor do
-          if registered_process = SolidQueue::Process.find_by(name: terminated_instance.name)
-            error = Processes::ThreadTerminatedError.new(terminated_instance.name)
-            registered_process.fail_all_claimed_executions_with(error)
-          end
-        end
-      end
+    def perform_immediate_termination
+      exit!
+    end
+
+    def all_processes_terminated?
+      process_instances.values.none?(&:alive?)
+    end
   end
 end

lib/solid_queue/fork_supervisor.rb

@@ -39,7 +39,8 @@ def reap_terminated_forks
         break unless pid
 
         if (terminated_fork = process_instances.delete(pid)) && !status.exited? || status.exitstatus > 0
-          handle_claimed_jobs_by(terminated_fork, status)
+          error = Processes::ProcessExitError.new(status)
+          release_claimed_jobs_by(terminated_fork, with_error: error)
         end
 
         configured_processes.delete(pid)
@@ -52,25 +53,14 @@ def replace_fork(pid, status)
       SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
         if terminated_fork = process_instances.delete(pid)
           payload[:fork] = terminated_fork
-          handle_claimed_jobs_by(terminated_fork, status)
+          error = Processes::ProcessExitError.new(status)
+          release_claimed_jobs_by(terminated_fork, with_error: error)
 
           start_process(configured_processes.delete(pid))
         end
       end
     end
 
-    # When a supervised fork crashes or exits we need to mark all the
-    # executions it had claimed as failed so that they can be retried
-    # by some other worker.
-    def handle_claimed_jobs_by(terminated_fork, status)
-      wrap_in_app_executor do
-        if registered_process = SolidQueue::Process.find_by(name: terminated_fork.name)
-          error = Processes::ProcessExitError.new(status)
-          registered_process.fail_all_claimed_executions_with(error)
-        end
-      end
-    end
-
     def all_processes_terminated?
       process_instances.empty?
     end

lib/solid_queue/supervisor.rb

@@ -144,4 +144,4 @@ def sync_std_streams
         STDOUT.sync = STDERR.sync = true
       end
   end
-end
+end

lib/solid_queue/supervisor/maintenance.rb

@@ -32,5 +32,16 @@ def fail_orphaned_executions
           ClaimedExecution.orphaned.fail_all_with(Processes::ProcessMissingError.new)
         end
       end
+
+      # When a supervised process crashes or exits we need to mark all the
+      # executions it had claimed as failed so that they can be retried
+      # by some other worker.
+      def release_claimed_jobs_by(terminated_process, with_error:)
+        wrap_in_app_executor do
+          if registered_process = SolidQueue::Process.find_by(name: terminated_process.name)
+            registered_process.fail_all_claimed_executions_with(with_error)
+          end
+        end
+      end
   end
 end

test/unit/fork_supervisor_test.rb

@@ -186,8 +186,8 @@ class ForkSupervisorTest < ActiveSupport::TestCase
   end
 
   # Regression test for supervisor failing to handle claimed jobs when its own
-  # process record has been pruned (NoMethodError in #handle_claimed_jobs_by).
-  test "handle_claimed_jobs_by fails claimed executions even if supervisor record is missing" do
+  # process record has been pruned (NoMethodError in #release_claimed_jobs_by).
+  test "release_claimed_jobs_by fails claimed executions even if supervisor record is missing" do
     worker_name = "worker-test-#{SecureRandom.hex(4)}"
 
     worker_process = SolidQueue::Process.register(kind: "Worker", pid: 999_999, name: worker_name)
@@ -196,20 +196,14 @@ class ForkSupervisorTest < ActiveSupport::TestCase
     claimed_execution = SolidQueue::ReadyExecution.claim("*", 1, worker_process.id).first
 
     terminated_fork = Struct.new(:name).new(worker_name)
-
-    DummyStatus = Struct.new(:pid, :exitstatus) do
-      def signaled? = false
-      def termsig = nil
-    end
-    status = DummyStatus.new(worker_process.pid, 1)
-
     supervisor = SolidQueue::ForkSupervisor.allocate
+    error = RuntimeError.new
 
-    supervisor.send(:handle_claimed_jobs_by, terminated_fork, status)
+    supervisor.send(:release_claimed_jobs_by, terminated_fork, with_error: error)
 
     failed = SolidQueue::FailedExecution.find_by(job_id: claimed_execution.job_id)
     assert failed.present?
-    assert_equal "SolidQueue::Processes::ProcessExitError", failed.exception_class
+    assert_equal "RuntimeError", failed.exception_class
   end
 
   private