Workers now yield self to lifecycle hooks

Author: Th3-M4jor

README.md

@@ -379,6 +379,9 @@ And into two different points in the worker's, dispatcher's and scheduler's life
 - `(worker|dispatcher|scheduler)_start`: after the worker/dispatcher/scheduler has finished booting and right before it starts the polling loop or loading the recurring schedule.
 - `(worker|dispatcher|scheduler)_stop`: after receiving a signal (`TERM`, `INT` or `QUIT`) and right before starting graceful or immediate shutdown (which is just `exit!`).
 
+The hooks for workers will have the worker instance yielded to the block so that you may read its configuration
+for logging or other metrics reporting purposes.
+
 You can use the following methods with a block to do this:
 ```ruby
 SolidQueue.on_start
@@ -398,6 +401,14 @@ For example:
 ```ruby
 SolidQueue.on_start { start_metrics_server }
 SolidQueue.on_stop { stop_metrics_server }
+
+SolidQueue.on_worker_start do |worker|
+  Rails.logger.info "Worker #{worker.name} started with queues: #{worker.queues.join(',')}"
+end
+
+SolidQueue.on_worker_stop do |worker|
+  Rails.logger.info "Worker #{worker.name} stopped with queues: #{worker.queues.join(',')}"
+end
 ```
 
 These can be called several times to add multiple hooks, but it needs to happen before Solid Queue is started. An initializer would be a good place to do this.

lib/solid_queue.rb

@@ -45,15 +45,15 @@ module SolidQueue
 
   [ Dispatcher, Scheduler, Worker ].each do |process|
     define_singleton_method(:"on_#{process.name.demodulize.downcase}_start") do |&block|
-      process.on_start { block.call }
+      process.on_start(&block)
     end
 
     define_singleton_method(:"on_#{process.name.demodulize.downcase}_stop") do |&block|
-      process.on_stop { block.call }
+      process.on_stop(&block)
     end
 
     define_singleton_method(:"on_#{process.name.demodulize.downcase}_exit") do |&block|
-      process.on_exit { block.call }
+      process.on_exit(&block)
     end
   end
 

lib/solid_queue/lifecycle_hooks.rb

@@ -43,10 +43,18 @@ def run_exit_hooks
 
       def run_hooks_for(event)
         self.class.lifecycle_hooks.fetch(event, []).each do |block|
-          block.call
+          if yield_self_to_hooks?
+            block.call(self)
+          else
+            block.call
+          end
         rescue Exception => exception
           handle_thread_error(exception)
         end
       end
+
+      def yield_self_to_hooks?
+        false
+      end
   end
 end

lib/solid_queue/worker.rb

@@ -8,12 +8,14 @@ class Worker < Processes::Poller
     before_shutdown :run_stop_hooks
     after_shutdown :run_exit_hooks
 
-    attr_accessor :queues, :pool
+    attr_reader :queues
 
     def initialize(**options)
       options = options.dup.with_defaults(SolidQueue::Configuration::WORKER_DEFAULTS)
 
-      @queues = Array(options[:queues])
+      # Ensure that the queues array is deep frozen to prevent accidental modification
+      @queues = Array(options[:queues]).map(&:freeze).freeze
+
       @pool = Pool.new(options[:threads], on_idle: -> { wake_up })
 
       super(**options)
@@ -24,6 +26,8 @@ def metadata
     end
 
     private
+      attr_reader :pool
+
       def poll
         claim_executions.then do |executions|
           executions.each do |execution|
@@ -54,5 +58,9 @@ def all_work_completed?
       def set_procline
         procline "waiting for jobs in #{queues.join(",")}"
       end
+
+      def yield_self_to_hooks?
+        true
+      end
   end
 end

test/integration/lifecycle_hooks_test.rb

@@ -10,9 +10,17 @@ class LifecycleHooksTest < ActiveSupport::TestCase
     SolidQueue.on_stop { JobResult.create!(status: :hook_called, value: :stop) }
     SolidQueue.on_exit { JobResult.create!(status: :hook_called, value: :exit) }
 
-    SolidQueue.on_worker_start { JobResult.create!(status: :hook_called, value: :worker_start) }
-    SolidQueue.on_worker_stop { JobResult.create!(status: :hook_called, value: :worker_stop) }
-    SolidQueue.on_worker_exit { JobResult.create!(status: :hook_called, value: :worker_exit) }
+    SolidQueue.on_worker_start do |w|
+      JobResult.create!(status: :hook_called, value: "worker_#{w.queues.join}_start")
+    end
+
+    SolidQueue.on_worker_stop do |w|
+      JobResult.create!(status: :hook_called, value: "worker_#{w.queues.join}_stop")
+    end
+
+    SolidQueue.on_worker_exit do |w|
+      JobResult.create!(status: :hook_called, value: "worker_#{w.queues.join}_exit")
+    end
 
     SolidQueue.on_dispatcher_start { JobResult.create!(status: :hook_called, value: :dispatcher_start) }
     SolidQueue.on_dispatcher_stop { JobResult.create!(status: :hook_called, value: :dispatcher_stop) }
@@ -22,23 +30,29 @@ class LifecycleHooksTest < ActiveSupport::TestCase
     SolidQueue.on_scheduler_stop { JobResult.create!(status: :hook_called, value: :scheduler_stop) }
     SolidQueue.on_scheduler_stop { JobResult.create!(status: :hook_called, value: :scheduler_exit) }
 
-    pid = run_supervisor_as_fork(workers: [ { queues: "*" } ], dispatchers: [ { batch_size: 100 } ], skip_recurring: false)
-    wait_for_registered_processes(4)
+    pid = run_supervisor_as_fork(
+      workers: [ { queues: "first_queue" }, { queues: "second_queue", processes: 1 } ],
+      dispatchers: [ { batch_size: 100 } ],
+      skip_recurring: false
+    )
+
+    wait_for_registered_processes(5)
 
     terminate_process(pid)
     wait_for_registered_processes(0)
 
 
     results = skip_active_record_query_cache do
       job_results = JobResult.where(status: :hook_called)
-      assert_equal 12, job_results.count
+      assert_equal 15, job_results.count
       job_results
     end
 
-    assert_equal({ "hook_called" => 12 }, results.map(&:status).tally)
+    assert_equal({ "hook_called" => 15 }, results.map(&:status).tally)
     assert_equal %w[
       start stop exit
-      worker_start worker_stop worker_exit
+      worker_first_queue_start worker_first_queue_stop worker_first_queue_exit
+      worker_second_queue_start worker_second_queue_stop worker_second_queue_exit
       dispatcher_start dispatcher_stop dispatcher_exit
       scheduler_start scheduler_stop scheduler_exit
     ].sort, results.map(&:value).sort

test/unit/worker_test.rb

@@ -156,7 +156,7 @@ class WorkerTest < ActiveSupport::TestCase
     @worker.start
     wait_for_registered_processes(1, timeout: 1.second)
 
-    assert_not @worker.pool.shutdown?
+    assert_not @worker.instance_variable_get(:@pool).shutdown?
 
     process = SolidQueue::Process.first
     assert_equal "Worker", process.kind
@@ -165,8 +165,8 @@ class WorkerTest < ActiveSupport::TestCase
 
     # And now just wait until the worker tries to heartbeat and realises
     # it needs to stop
-    wait_while_with_timeout(2) { !@worker.pool.shutdown? }
-    assert @worker.pool.shutdown?
+    wait_while_with_timeout(2) { !@worker.instance_variable_get(:@pool).shutdown? }
+    assert @worker.instance_variable_get(:@pool).shutdown?
   ensure
     SolidQueue.process_heartbeat_interval = old_heartbeat_interval
   end