Fix worker pool defaults and async shutdown

Author: crmne

lib/solid_queue/execution_pools/async_pool.rb

@@ -5,6 +5,8 @@ module ExecutionPools
     class AsyncPool
       include AppExecutor
 
+      SHUTDOWN_SENTINEL = Object.new
+
       class MissingDependencyError < LoadError
         def initialize(error)
           super(
@@ -91,13 +93,13 @@ def idle?
       end
 
       def shutdown
-        should_close = state_mutex.synchronize do
+        should_enqueue_shutdown = state_mutex.synchronize do
           next false if @shutdown
 
           @shutdown = true
         end
 
-        queue.close if should_close
+        queue.enqueue(SHUTDOWN_SENTINEL) if should_enqueue_shutdown
       end
 
       def shutdown?
@@ -141,6 +143,8 @@ def start_reactor
         def drain_queue(task, semaphore)
           task.async do
             while execution = queue.dequeue
+              break if execution.equal?(SHUTDOWN_SENTINEL)
+
               semaphore.async(execution) do |_execution_task, scheduled_execution|
                 perform_execution(scheduled_execution)
               end

lib/solid_queue/worker.rb

@@ -12,7 +12,7 @@ class Worker < Processes::Poller
 
     def initialize(**options)
       options = options.dup
-      options[:threads] = options[:capacity] || options[:fibers] || options[:threads]
+      options[:threads] = options[:capacity] || options[:fibers] if options.key?(:capacity) || options.key?(:fibers)
       options = options.with_defaults(SolidQueue::Configuration::WORKER_DEFAULTS)
 
       # Ensure that the queues array is deep frozen to prevent accidental modification

test/unit/execution_pools/async_pool_test.rb

@@ -82,6 +82,21 @@ def test_waits_for_in_flight_executions_during_shutdown
     end
   end
 
+  def test_shutdown_does_not_depend_on_async_queue_close
+    with_execution_isolation(:fiber) do
+      pool = SolidQueue::ExecutionPools::AsyncPool.new(1)
+      queue = pool.instance_variable_get(:@queue)
+      queue.define_singleton_method(:close) { raise "should not be called" }
+
+      pool.shutdown
+
+      assert pool.wait_for_termination(1.second)
+    ensure
+      pool&.shutdown
+      pool&.wait_for_termination(1.second)
+    end
+  end
+
   def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled
     with_execution_isolation(:fiber) do
       notifications = Thread::Queue.new

test/unit/worker_test.rb

@@ -123,6 +123,19 @@ class WorkerTest < ActiveSupport::TestCase
     wait_for_registered_processes(0, timeout: 1.second)
   end
 
+  test "defaults thread workers to the configured thread pool size" do
+    worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.2)
+
+    worker.start
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    assert_equal 3, worker.pool.size
+    assert_metadata SolidQueue::Process.first, thread_pool_size: 3, capacity: 3, execution_mode: "thread"
+  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