Extract termination instrumentation to Supervisor

Author: rosa

lib/solid_queue/supervisor.rb

@@ -63,6 +63,10 @@ def start_process(configured_process)
         raise NotImplementedError
       end
 
+      def instrument_termination(type, &block)
+        SolidQueue.instrument("#{type}_termination".to_sym, process_id: process_id, supervisor_pid: ::Process.pid, &block)
+      end
+
       def shutdown
         SolidQueue.instrument(:shutdown_process, process: self) do
           run_callbacks(:shutdown) do

lib/solid_queue/supervisor/async_supervisor.rb

@@ -62,7 +62,9 @@ def stop_threads
       end
 
       def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: threads.keys) do |payload|
+        instrument_termination(:graceful) do |payload|
+          payload[:supervised_processes] = threads.keys
+
           unless all_threads_terminated?
             payload[:shutdown_timeout_exceeded] = true
             terminate_immediately
@@ -71,7 +73,9 @@ def terminate_gracefully
       end
 
       def terminate_immediately
-        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: threads.keys) do
+        instrument_termination(:immediate) do |payload|
+          payload[:supervised_processes] = threads.keys
+
           exit!
         end
       end

lib/solid_queue/supervisor/fork_supervisor.rb

@@ -47,7 +47,9 @@ def start_process(configured_process)
       end
 
       def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do |payload|
+        instrument_termination(:graceful) do |payload|
+          payload[:supervised_processes] = forks.keys
+
           term_forks
 
           Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
@@ -62,7 +64,9 @@ def terminate_gracefully
       end
 
       def terminate_immediately
-        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do
+        instrument_termination(:immediate) do |payload|
+          payload[:supervised_processes] = forks.keys
+
           quit_forks
         end
       end

test/integration/async_processes_lifecycle_test.rb

@@ -33,9 +33,9 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase
 
   test "kill supervisor while there are jobs in-flight" do
     no_pause = enqueue_store_result_job("no pause")
-    pause = enqueue_store_result_job("pause", pause: 0.2.seconds)
+    pause = enqueue_store_result_job("pause", pause: 1.second)
 
-    signal_process(@pid, :KILL, wait: 0.15.seconds)
+    signal_process(@pid, :KILL, wait: 0.2.seconds)
     wait_for_jobs_to_finish_for(2.seconds)
     wait_for_registered_processes(1, timeout: 3.second)
 

test/integration/forked_processes_lifecycle_test.rb

@@ -33,9 +33,9 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
 
   test "kill supervisor while there are jobs in-flight" do
     no_pause = enqueue_store_result_job("no pause")
-    pause = enqueue_store_result_job("pause", pause: 0.2.seconds)
+    pause = enqueue_store_result_job("pause", pause: 1.second)
 
-    signal_process(@pid, :KILL, wait: 0.15.seconds)
+    signal_process(@pid, :KILL, wait: 0.2.seconds)
     wait_for_jobs_to_finish_for(2.seconds)
     wait_for_registered_processes(1, timeout: 3.second)