Remove mode option from configuration and rely on the config values

That is, no more "work" or "dispatch". The behaviour is the same as "all",
and to get work only or dispatch only it is enough with passing an empty
array for dispatchers and workers.

Author: rosa

lib/puma/plugin/solid_queue.rb

@@ -14,7 +14,7 @@ def start(launcher)
     launcher.events.on_booted do
       @solid_queue_pid = fork do
         Thread.new { monitor_puma }
-        SolidQueue::Supervisor.start(mode: :all)
+        SolidQueue::Supervisor.start
       end
     end
 

lib/solid_queue/configuration.rb

@@ -17,38 +17,25 @@ class Configuration
       recurring_tasks: []
     }
 
-    def initialize(mode: :work, load_from: nil)
-      @mode = mode
+    def initialize(load_from: nil)
       @raw_config = config_from(load_from)
     end
 
     def processes
-      case mode
-      when :dispatch then dispatchers
-      when :work     then workers
-      when :all      then dispatchers + workers
-      else           raise "Invalid mode #{mode}"
-      end
+      dispatchers + workers
     end
 
     def workers
-      if mode.in? %i[ work all]
-        workers_options.flat_map do |worker_options|
-          processes = worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
-          processes.times.map { Worker.new(**worker_options.with_defaults(WORKER_DEFAULTS)) }
-        end
-      else
-        []
+      workers_options.flat_map do |worker_options|
+        processes = worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
+        processes.times.map { Worker.new(**worker_options.with_defaults(WORKER_DEFAULTS)) }
       end
     end
 
     def dispatchers
-      if mode.in? %i[ dispatch all]
-        dispatchers_options.map do |dispatcher_options|
-          recurring_tasks = parse_recurring_tasks dispatcher_options[:recurring_tasks]
-
-          Dispatcher.new **dispatcher_options.merge(recurring_tasks: recurring_tasks).with_defaults(DISPATCHER_DEFAULTS)
-        end
+      dispatchers_options.map do |dispatcher_options|
+        recurring_tasks = parse_recurring_tasks dispatcher_options[:recurring_tasks]
+        Dispatcher.new **dispatcher_options.merge(recurring_tasks: recurring_tasks).with_defaults(DISPATCHER_DEFAULTS)
       end
     end
 
@@ -58,7 +45,7 @@ def max_number_of_threads
     end
 
     private
-      attr_reader :raw_config, :mode
+      attr_reader :raw_config
 
       DEFAULT_CONFIG_FILE_PATH = "config/solid_queue.yml"
 

lib/solid_queue/supervisor.rb

@@ -5,9 +5,9 @@ class Supervisor < Processes::Base
     include Processes::Signals
 
     class << self
-      def start(mode: :work, load_configuration_from: nil)
+      def start(mode: :fork, load_configuration_from: nil)
         SolidQueue.supervisor = true
-        configuration = Configuration.new(mode: mode, load_from: load_configuration_from)
+        configuration = Configuration.new(load_from: load_configuration_from)
 
         new(*configuration.processes).start
       end

lib/solid_queue/tasks.rb

@@ -1,16 +1,6 @@
 namespace :solid_queue do
   desc "start solid_queue supervisor to dispatch and process jobs"
   task start: :environment do
-    SolidQueue::Supervisor.start(mode: :all)
-  end
-
-  desc "start solid_queue supervisor to process jobs"
-  task work: :environment do
-    SolidQueue::Supervisor.start(mode: :work)
-  end
-
-  desc "start solid_queue dispatcher to enqueue scheduled jobs"
-  task dispatch: :environment do
-    SolidQueue::Supervisor.start(mode: :dispatch)
+    SolidQueue::Supervisor.start
   end
 end

test/integration/concurrency_controls_test.rb

@@ -11,7 +11,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     default_worker = { queues: "default", polling_interval: 0.1, processes: 3, threads: 2 }
     dispatcher = { polling_interval: 0.1, batch_size: 200, concurrency_maintenance_interval: 1 }
 
-    @pid = run_supervisor_as_fork(mode: :all, load_configuration_from: { workers: [ default_worker ], dispatchers: [ dispatcher ] })
+    @pid = run_supervisor_as_fork(load_configuration_from: { workers: [ default_worker ], dispatchers: [ dispatcher ] })
 
     wait_for_registered_processes(5, timeout: 0.5.second) # 3 workers working the default queue + dispatcher + supervisor
   end

test/integration/processes_lifecycle_test.rb

@@ -6,7 +6,8 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
-    @pid = run_supervisor_as_fork
+    config_as_hash = { workers: [ { queues: :background }, { queues: :default, threads: 5 } ], dispatchers: [] }
+    @pid = run_supervisor_as_fork(load_configuration_from: config_as_hash)
 
     wait_for_registered_processes(3, timeout: 3.second)
     assert_registered_workers_for(:background, :default)
@@ -17,6 +18,7 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
 
     SolidQueue::Process.destroy_all
     SolidQueue::Job.destroy_all
+    JobResult.delete_all
   end
 
   test "enqueue jobs in multiple queues" do
@@ -142,13 +144,13 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
   end
 
   test "process some jobs that raise errors" do
-    enqueue_store_result_job("no error", :background, 2)
-    enqueue_store_result_job("no error", :default, 2)
-    error1 = enqueue_store_result_job("error", :background, 1, exception: RuntimeError)
-    enqueue_store_result_job("no error", :background, 1, pause: 0.03)
-    error2 = enqueue_store_result_job("error", :background, 1, exception: RuntimeError, pause: 0.05)
-    enqueue_store_result_job("no error", :default, 2, pause: 0.01)
-    error3 = enqueue_store_result_job("error", :default, 1, exception: RuntimeError)
+    2.times { enqueue_store_result_job("no error", :background) }
+    2.times { enqueue_store_result_job("no error", :default) }
+    error1 = enqueue_store_result_job("error", :background, exception: RuntimeError)
+    enqueue_store_result_job("no error", :background, pause: 0.03)
+    error2 = enqueue_store_result_job("error", :background, exception: RuntimeError, pause: 0.05)
+    2.times { enqueue_store_result_job("no error", :default, pause: 0.01) }
+    error3 = enqueue_store_result_job("error", :default, exception: RuntimeError)
 
     wait_for_jobs_to_finish_for(0.5.seconds)
 
@@ -165,12 +167,15 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
   end
 
   test "process a job that exits" do
-    enqueue_store_result_job("no exit", :background, 2)
-    enqueue_store_result_job("no exit", :default, 2)
-    enqueue_store_result_job("paused no exit", :default, 1, pause: 0.5)
-    exit_job = enqueue_store_result_job("exit", :background, 1, exit: true, pause: 0.2)
-    pause_job = enqueue_store_result_job("exit", :background, 1, pause: 0.3)
-    enqueue_store_result_job("no exit", :background, 2)
+    2.times do
+      enqueue_store_result_job("no exit", :background)
+      enqueue_store_result_job("no exit", :default)
+    end
+    enqueue_store_result_job("paused no exit", :default, pause: 0.5)
+    exit_job = enqueue_store_result_job("exit", :background, exit: true, pause: 0.2)
+    pause_job = enqueue_store_result_job("exit", :background, pause: 0.3)
+
+    2.times { enqueue_store_result_job("no exit", :background) }
 
     wait_for_jobs_to_finish_for(5.seconds)
 
@@ -215,10 +220,8 @@ def assert_no_registered_workers
       assert_empty find_processes_registered_as("Worker").to_a
     end
 
-    def enqueue_store_result_job(value, queue_name = :background, count = 1, **options)
-      count.times.collect { StoreResultJob.set(queue: queue_name).perform_later(value, **options) }.then do |jobs|
-        jobs.one? ? jobs.first : jobs
-      end
+    def enqueue_store_result_job(value, queue_name = :background, **options)
+      StoreResultJob.set(queue: queue_name).perform_later(value, **options)
     end
 
     def assert_completed_job_results(value, queue_name = :background, count = 1)

test/integration/recurring_tasks_test.rb

@@ -6,7 +6,7 @@ class RecurringTasksTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
-    @pid = run_supervisor_as_fork(mode: :all)
+    @pid = run_supervisor_as_fork
     # 1 supervisor + 2 workers + 1 dispatcher
     wait_for_registered_processes(4, timeout: 3.second)
   end

test/unit/configuration_test.rb

@@ -3,7 +3,7 @@
 class ConfigurationTest < ActiveSupport::TestCase
   test "default configuration to process all queues and dispatch" do
     configuration = stub_const(SolidQueue::Configuration, :DEFAULT_CONFIG_FILE_PATH, "non/existent/path") do
-      SolidQueue::Configuration.new(mode: :all)
+      SolidQueue::Configuration.new
     end
 
     assert_equal 2, configuration.processes.count
@@ -16,7 +16,7 @@ class ConfigurationTest < ActiveSupport::TestCase
   end
 
   test "read configuration from default file" do
-    configuration = SolidQueue::Configuration.new(mode: :all)
+    configuration = SolidQueue::Configuration.new
     assert 3, configuration.processes.count
     assert_equal 2, configuration.workers.count
     assert_equal 1, configuration.dispatchers.count
@@ -26,7 +26,7 @@ class ConfigurationTest < ActiveSupport::TestCase
     background_worker = { queues: "background", polling_interval: 10 }
     dispatcher = { batch_size: 100 }
     config_as_hash = { workers: [ background_worker, background_worker ], dispatchers: [ dispatcher ] }
-    configuration = SolidQueue::Configuration.new(mode: :all, load_from: config_as_hash)
+    configuration = SolidQueue::Configuration.new(load_from: config_as_hash)
 
     assert_equal 1, configuration.dispatchers.count
     dispatcher = configuration.dispatchers.first
@@ -39,14 +39,14 @@ class ConfigurationTest < ActiveSupport::TestCase
   end
 
   test "max number of threads" do
-    configuration = SolidQueue::Configuration.new(mode: :all)
+    configuration = SolidQueue::Configuration.new
     assert 7, configuration.max_number_of_threads
   end
 
   test "mulitple workers with the same configuration" do
     background_worker = { queues: "background", polling_interval: 10, processes: 3 }
     config_as_hash = { workers: [ background_worker ] }
-    configuration = SolidQueue::Configuration.new(mode: :work, load_from: config_as_hash)
+    configuration = SolidQueue::Configuration.new(load_from: config_as_hash)
 
     assert_equal 3, configuration.workers.count
     assert_equal [ "background" ], configuration.workers.flat_map(&:queues).uniq

test/unit/supervisor_test.rb

@@ -16,23 +16,26 @@ class SupervisorTest < ActiveSupport::TestCase
     SolidQueue::Process.destroy_all
   end
 
-  test "start in work mode (default)" do
+  test "start" do
     pid = run_supervisor_as_fork
     wait_for_registered_processes(4)
 
     assert_registered_supervisor(pid)
     assert_registered_workers(2, supervisor_pid: pid)
+    assert_registered_dispatcher(supervisor_pid: pid)
 
     terminate_process(pid)
 
     assert_no_registered_processes
   end
 
-  test "start in dispatch mode" do
-    pid = run_supervisor_as_fork(mode: :dispatch)
-    wait_for_registered_processes(4)
+  test "start with provided configuration" do
+    config_as_hash = { workers: [], dispatchers: [ { batch_size: 100 } ] }
+    pid = run_supervisor_as_fork(load_configuration_from: config_as_hash)
+    wait_for_registered_processes(2) # supervisor + dispatcher
 
     assert_registered_supervisor(pid)
+    assert_registered_workers(0)
     assert_registered_dispatcher(supervisor_pid: pid)
 
     terminate_process(pid)
@@ -114,7 +117,7 @@ class SupervisorTest < ActiveSupport::TestCase
   end
 
   private
-    def assert_registered_workers(count, supervisor_pid:, **metadata)
+    def assert_registered_workers(count, supervisor_pid: nil, **metadata)
       skip_active_record_query_cache do
         workers = find_processes_registered_as("Worker")
         assert_equal count, workers.count