Fail fast for async worker config errors

Author: crmne

lib/solid_queue/configuration.rb

@@ -8,6 +8,8 @@ class Configuration
     validate :ensure_valid_recurring_tasks
     validate :ensure_correctly_sized_thread_pool
     validate :ensure_valid_worker_execution_modes
+    validate :ensure_async_workers_use_capacity_aliases
+    validate :ensure_async_workers_have_required_dependency
     validate :ensure_async_workers_use_supported_isolation_level
 
     class Process < Struct.new(:kind, :attributes)
@@ -106,6 +108,22 @@ def ensure_valid_worker_execution_modes
         end
       end
 
+      def ensure_async_workers_use_capacity_aliases
+        workers_options.each do |options|
+          if async_worker?(options) && options.key?(:threads)
+            errors.add(:base, "Async workers do not accept `threads`. Use `capacity` or `fibers` instead.")
+          end
+        end
+      end
+
+      def ensure_async_workers_have_required_dependency
+        return unless workers_options.any? { |options| async_worker?(options) }
+
+        SolidQueue::ExecutionPools::AsyncPool.ensure_dependency!
+      rescue LoadError => error
+        errors.add(:base, error.message)
+      end
+
       def ensure_async_workers_use_supported_isolation_level
         return unless workers_options.any? { |options| async_worker?(options) }
 
@@ -281,7 +299,9 @@ def async_worker?(options)
 
       def worker_defaults_for(options)
         if async_worker?(options)
-          WORKER_DEFAULTS.except(:threads).merge(capacity: WORKER_DEFAULTS[:threads])
+          WORKER_DEFAULTS.except(:threads).tap do |defaults|
+            defaults[:capacity] = WORKER_DEFAULTS[:threads] unless options.key?(:threads)
+          end
         else
           WORKER_DEFAULTS
         end

lib/solid_queue/execution_pools/async_pool.rb

@@ -27,6 +27,14 @@ def initialize(level)
       end
 
       class << self
+        def ensure_dependency!
+          require "async"
+          require "async/queue"
+          require "async/semaphore"
+        rescue LoadError => error
+          raise MissingDependencyError.new(error)
+        end
+
         def ensure_supported_isolation_level!
           return if supported_isolation_level?
 
@@ -50,7 +58,7 @@ def initialize(size, on_state_change: nil)
         @fatal_error = nil
         @boot_queue = Thread::Queue.new
 
-        load_dependency!
+        self.class.ensure_dependency!
         self.class.ensure_supported_isolation_level!
 
         @queue = Async::Queue.new
@@ -115,14 +123,6 @@ def name
           @name ||= "solid_queue-async-pool-#{object_id}"
         end
 
-        def load_dependency!
-          require "async"
-          require "async/queue"
-          require "async/semaphore"
-        rescue LoadError => error
-          raise MissingDependencyError.new(error)
-        end
-
         def start_reactor
           create_thread do
             Async do |task|

test/unit/configuration_test.rb

@@ -77,6 +77,19 @@ class ConfigurationTest < ActiveSupport::TestCase
     end
   end
 
+  test "async workers reject threads in favor of capacity aliases" do
+    with_execution_isolation(:fiber) do
+      configuration = SolidQueue::Configuration.new(
+        workers: [ { queues: "llm*", execution_mode: :async, threads: 10 } ],
+        dispatchers: [],
+        skip_recurring: true
+      )
+
+      assert_not configuration.valid?
+      assert_match /Async workers do not accept `threads`/, configuration.errors.full_messages.first
+    end
+  end
+
   test "async worker capacity does not inflate required database pool size" do
     with_execution_isolation(:fiber) do
       configuration = SolidQueue::Configuration.new(
@@ -201,6 +214,16 @@ class ConfigurationTest < ActiveSupport::TestCase
     assert_not configuration.valid?
     assert_match /requires fiber-scoped isolated execution state/, configuration.errors.full_messages.first
 
+    with_execution_isolation(:fiber) do
+      load_error = LoadError.new("cannot load such file -- async")
+      missing_dependency_error = SolidQueue::ExecutionPools::AsyncPool::MissingDependencyError.new(load_error)
+      SolidQueue::ExecutionPools::AsyncPool.expects(:ensure_dependency!).raises(missing_dependency_error)
+
+      configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { execution_mode: :async } ])
+      assert_not configuration.valid?
+      assert_match /gem "async"/, configuration.errors.full_messages.first
+    end
+
     # Not enough DB connections
     configuration = SolidQueue::Configuration.new(workers: [ { queues: "background", threads: 50, polling_interval: 10 } ])
     assert_not configuration.valid?

test/unit/execution_pools/async_pool_test.rb

@@ -19,7 +19,7 @@ def perform
   def test_raises_a_clear_error_when_the_async_gem_is_unavailable
     load_error = LoadError.new("cannot load such file -- async")
 
-    SolidQueue::ExecutionPools::AsyncPool.any_instance.expects(:require).with("async").raises(load_error)
+    SolidQueue::ExecutionPools::AsyncPool.expects(:require).with("async").raises(load_error)
 
     error = assert_raises SolidQueue::ExecutionPools::AsyncPool::MissingDependencyError do
       SolidQueue::ExecutionPools::AsyncPool.new(3)