Build worker pools after fork

Author: crmne

lib/solid_queue/worker.rb

@@ -18,11 +18,11 @@ def initialize(**options)
       # Ensure that the queues array is deep frozen to prevent accidental modification
       @queues = Array(options[:queues]).map(&:freeze).freeze
 
-      @pool = ExecutionPools.build(
+      @pool_options = {
         mode: options[:execution_mode],
         size: options[:threads],
         on_state_change: -> { wake_up }
-      )
+      }
 
       super(**options)
     end
@@ -50,6 +50,11 @@ def claim_executions
         end
       end
 
+      def boot
+        build_pool
+        super
+      end
+
       def shutdown
         pool.shutdown
         pool.wait_for_termination(SolidQueue.shutdown_timeout)
@@ -69,5 +74,9 @@ def heartbeat
       def set_procline
         procline "waiting for jobs in #{queues.join(",")}"
       end
+
+      def build_pool
+        @pool ||= ExecutionPools.build(**@pool_options)
+      end
   end
 end

test/unit/worker_test.rb

@@ -30,6 +30,26 @@ class WorkerTest < ActiveSupport::TestCase
     }
   end
 
+  test "builds the execution pool on boot instead of initialize" do
+    pool = SolidQueue::ExecutionPools::ThreadPool.new(3)
+
+    SolidQueue::ExecutionPools.expects(:build).once.with do |**options|
+      options[:mode] == :thread && options[:size] == 3 && options[:on_state_change].respond_to?(:call)
+    end.returns(pool)
+
+    worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.2)
+
+    assert_nil worker.pool
+
+    worker.start
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    assert_equal pool, worker.pool
+  ensure
+    worker&.stop
+    wait_for_registered_processes(0, timeout: 1.second)
+  end
+
   test "errors on polling are passed to on_thread_error and re-raised" do
     errors = Concurrent::Array.new