Split recurring scheduling responsibilities into another process

rosa · rosa · commit a4467dbe57b2 · 2024-09-09T19:22:23.000+02:00
A Scheduler, instead of Dispatcher, that will be created only when there
are recurring tasks to run.
diff --git a/lib/solid_queue/cli.rb b/lib/solid_queue/cli.rb
@@ -14,7 +14,6 @@ class Cli < Thor
       desc: "Path to recurring schedule definition",
       banner: "SOLID_QUEUE_RECURRING_SCHEDULE"
 
-    class_option :dispatch_only, type: :boolean, default: false
     class_option :work_only, type: :boolean, default: false
     class_option :skip_recurring, type: :boolean, default: false
 
diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb
@@ -30,11 +30,9 @@ def initialize(**options)
     end
 
     def configured_processes
-      case
-      when only_work?     then workers
-      when only_dispatch? then dispatchers
+      if only_work? then workers
       else
-        dispatchers + workers
+        dispatchers + workers + schedulers
       end
     end
 
@@ -81,6 +79,14 @@ def dispatchers
         end
       end
 
+      def schedulers
+        if !skip_recurring_tasks? && recurring_tasks.any?
+          [ Process.new(:scheduler, recurring_tasks: recurring_tasks) ]
+        else
+          []
+        end
+      end
+
       def workers_options
         @workers_options ||= processes_config.fetch(:workers, [])
           .map { |options| options.dup.symbolize_keys }
@@ -89,22 +95,6 @@ def workers_options
       def dispatchers_options
         @dispatchers_options ||= processes_config.fetch(:dispatchers, [])
           .map { |options| options.dup.symbolize_keys }
-          .then { |options| with_recurring_tasks(options) }
-      end
-
-      def with_recurring_tasks(options)
-        if !skip_recurring_tasks? && recurring_tasks.any?
-          options.sort_by! { |attrs| attrs[:polling_interval] }
-
-          if least_busy_dispatcher = options.pop
-            least_busy_dispatcher[:recurring_tasks] = recurring_tasks
-            options.push(least_busy_dispatcher)
-          else
-            [ DISPATCHER_DEFAULTS.merge(recurring_tasks: recurring_tasks) ]
-          end
-        else
-          options
-        end
       end
 
       def recurring_tasks
@@ -128,7 +118,7 @@ def recurring_tasks_config
       def config_from(file_or_hash, keys: [], fallback: {}, env: Rails.env)
         load_config_from(file_or_hash).then do |config|
           config = config[env.to_sym] ? config[env.to_sym] : config
-          config = config.slice(*keys) if keys.any?
+          config = config.slice(*keys) if keys.any? && config.present?
 
           if config.empty? then fallback
           else
diff --git a/lib/solid_queue/dispatcher.rb b/lib/solid_queue/dispatcher.rb
@@ -2,24 +2,23 @@
 
 module SolidQueue
   class Dispatcher < Processes::Poller
-    attr_accessor :batch_size, :concurrency_maintenance, :recurring_schedule
+    attr_accessor :batch_size, :concurrency_maintenance
 
-    after_boot :start_concurrency_maintenance, :schedule_recurring_tasks
-    before_shutdown :stop_concurrency_maintenance, :unschedule_recurring_tasks
+    after_boot :start_concurrency_maintenance
+    before_shutdown :stop_concurrency_maintenance
 
     def initialize(**options)
       options = options.dup.with_defaults(SolidQueue::Configuration::DISPATCHER_DEFAULTS)
 
       @batch_size = options[:batch_size]
 
       @concurrency_maintenance = ConcurrencyMaintenance.new(options[:concurrency_maintenance_interval], options[:batch_size]) if options[:concurrency_maintenance]
-      @recurring_schedule = RecurringSchedule.new(options[:recurring_tasks])
 
       super(**options)
     end
 
     def metadata
-      super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval, recurring_schedule: recurring_schedule.task_keys.presence)
+      super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval)
     end
 
     private
diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb
@@ -94,7 +94,7 @@ def register_process(event)
     if error = event.payload[:error]
       warn formatted_event(event, action: "Error registering #{process_kind}", **attributes.merge(error: formatted_error(error)))
     else
-      info formatted_event(event, action: "Register #{process_kind}", **attributes)
+      debug formatted_event(event, action: "Register #{process_kind}", **attributes)
     end
   end
 
@@ -114,7 +114,7 @@ def deregister_process(event)
     if error = event.payload[:error]
       warn formatted_event(event, action: "Error deregistering #{process.kind}", **attributes.merge(error: formatted_error(error)))
     else
-      info formatted_event(event, action: "Deregister #{process.kind}", **attributes)
+      debug formatted_event(event, action: "Deregister #{process.kind}", **attributes)
     end
   end
 
diff --git a/lib/solid_queue/processes/poller.rb b/lib/solid_queue/processes/poller.rb
@@ -41,9 +41,6 @@ def poll
         raise NotImplementedError
       end
 
-      def shutdown
-      end
-
       def with_polling_volume
         SolidQueue.instrument(:polling) do
           if SolidQueue.silence_polling? && ActiveRecord::Base.logger
diff --git a/lib/solid_queue/processes/runnable.rb b/lib/solid_queue/processes/runnable.rb
@@ -18,6 +18,8 @@ def start
 
     def stop
       @stopped = true
+      wake_up if running_async?
+
       @thread&.join
     end
 
@@ -61,6 +63,9 @@ def all_work_completed?
         false
       end
 
+      def shutdown
+      end
+
       def set_procline
       end
 
diff --git a/lib/solid_queue/scheduler.rb b/lib/solid_queue/scheduler.rb
@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Scheduler < Processes::Base
+    include Processes::Runnable
+
+    attr_accessor :recurring_schedule
+
+    after_boot :schedule_recurring_tasks
+    before_shutdown :unschedule_recurring_tasks
+
+    def initialize(recurring_tasks:, **options)
+      @recurring_schedule = RecurringSchedule.new(recurring_tasks)
+
+      super(**options)
+    end
+
+    def metadata
+      super.merge(recurring_schedule: recurring_schedule.task_keys.presence)
+    end
+
+    private
+      SLEEP_INTERVAL = 300 # Right now it doesn't matter, can be set to 1 in the future for dynamic tasks
+
+      def run
+        loop do
+          break if shutting_down?
+
+          interruptible_sleep(SLEEP_INTERVAL)
+        end
+      ensure
+        SolidQueue.instrument(:shutdown_process, process: self) do
+          run_callbacks(:shutdown) { shutdown }
+        end
+      end
+
+      def schedule_recurring_tasks
+        recurring_schedule.schedule_tasks
+      end
+
+      def unschedule_recurring_tasks
+        recurring_schedule.unschedule_tasks
+      end
+
+      def all_work_completed?
+        recurring_schedule.empty?
+      end
+
+      def set_procline
+        procline "scheduling #{recurring_schedule.task_keys.join(",")}"
+      end
+  end
+end
diff --git a/lib/solid_queue/scheduler/recurring_schedule.rb b/lib/solid_queue/scheduler/recurring_schedule.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module SolidQueue
-  class Dispatcher::RecurringSchedule
+  class Scheduler::RecurringSchedule
     include AppExecutor
 
     attr_reader :configured_tasks, :scheduled_tasks
diff --git a/test/dummy/config/empty_configuration.yml b/test/dummy/config/empty_configuration.yml
@@ -1,7 +1,3 @@
-default: &default
+development: []
 
-development:
-  <<: *default
-
-test:
-  <<: *default
+test: []
diff --git a/test/integration/recurring_tasks_test.rb b/test/integration/recurring_tasks_test.rb
@@ -6,9 +6,9 @@ class RecurringTasksTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
-    @pid = run_supervisor_as_fork
-    # 1 supervisor + 2 workers + 1 dispatcher
-    wait_for_registered_processes(4, timeout: 3.second)
+    @pid = run_supervisor_as_fork(skip_recurring: false)
+    # 1 supervisor + 2 workers + 1 dispatcher + 1 scheduler
+    wait_for_registered_processes(5, timeout: 3.second)
   end
 
   teardown do
@@ -52,29 +52,29 @@ class RecurringTasksTest < ActiveSupport::TestCase
     task = SolidQueue::RecurringTask.find_by(key: "periodic_store_result")
     task.update!(class_name: "StoreResultJob", schedule: "every minute", arguments: [ 42 ])
 
-    @pid = run_supervisor_as_fork
-    wait_for_registered_processes(4, timeout: 3.second)
+    @pid = run_supervisor_as_fork(skip_recurring: false)
+    wait_for_registered_processes(5, timeout: 3.second)
 
     # Wait for concurrency schedule loading after process registration
     sleep(0.5)
 
     assert_recurring_tasks configured_task
 
     another_task = { example_task: { class: "AddToBufferJob", schedule: "every hour", args: [ 42 ] } }
-    dispatcher1 = SolidQueue::Dispatcher.new(concurrency_maintenance: false, recurring_tasks: another_task).tap(&:start)
-    wait_for_registered_processes(5, timeout: 1.second)
+    scheduler1 = SolidQueue::Scheduler.new(recurring_tasks: another_task).tap(&:start)
+    wait_for_registered_processes(6, timeout: 1.second)
 
     assert_recurring_tasks configured_task.merge(another_task)
 
     updated_task = { example_task: { class: "AddToBufferJob", schedule: "every minute" } }
-    dispatcher2 = SolidQueue::Dispatcher.new(concurrency_maintenance: false, recurring_tasks: updated_task).tap(&:start)
-    wait_for_registered_processes(6, timeout: 1.second)
+    scheduler2 = SolidQueue::Scheduler.new(recurring_tasks: updated_task).tap(&:start)
+    wait_for_registered_processes(7, timeout: 1.second)
 
     assert_recurring_tasks configured_task.merge(updated_task)
 
     terminate_process(@pid)
-    dispatcher1.stop
-    dispatcher2.stop
+    scheduler1.stop
+    scheduler2.stop
   end
 
   private
diff --git a/test/test_helpers/processes_test_helper.rb b/test/test_helpers/processes_test_helper.rb
@@ -3,7 +3,7 @@ module ProcessesTestHelper
 
   def run_supervisor_as_fork(**options)
     fork do
-      SolidQueue::Supervisor.start(**options)
+      SolidQueue::Supervisor.start(**options.with_defaults(skip_recurring: true))
     end
   end
 
@@ -38,6 +38,12 @@ def assert_registered_processes(kind:, count: 1, supervisor_pid: nil, **attribut
     end
   end
 
+  def assert_metadata(process, metadata)
+    metadata.each do |attr, value|
+      assert_equal value, process.metadata[attr.to_s]
+    end
+  end
+
   def find_processes_registered_as(kind)
     skip_active_record_query_cache do
       SolidQueue::Process.where(kind: kind)
diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb
@@ -4,21 +4,22 @@ class ConfigurationTest < ActiveSupport::TestCase
   test "default configuration to process all queues and dispatch" do
     configuration = SolidQueue::Configuration.new(config_file: nil)
 
-    assert_equal 2, configuration.configured_processes.count
+    assert_equal 3, configuration.configured_processes.count
     assert_processes configuration, :worker, 1, queues: "*"
     assert_processes configuration, :dispatcher, 1, batch_size: SolidQueue::Configuration::DISPATCHER_DEFAULTS[:batch_size]
+    assert_processes configuration, :scheduler, 1
   end
 
   test "default configuration when config given doesn't include any configuration" do
-    configuration = SolidQueue::Configuration.new(config_file: config_file_path(:invalid_configuration))
+    configuration = SolidQueue::Configuration.new(config_file: config_file_path(:invalid_configuration), skip_recurring: true)
 
     assert_equal 2, configuration.configured_processes.count
     assert_processes configuration, :worker, 1, queues: "*"
     assert_processes configuration, :dispatcher, 1, batch_size: SolidQueue::Configuration::DISPATCHER_DEFAULTS[:batch_size]
   end
 
   test "default configuration when config given is empty" do
-    configuration = SolidQueue::Configuration.new(config_file: config_file_path(:empty_configuration))
+    configuration = SolidQueue::Configuration.new(config_file: config_file_path(:empty_configuration), recurring_schedule_file: config_file_path(:empty_configuration))
 
     assert_equal 2, configuration.configured_processes.count
     assert_processes configuration, :worker, 1, queues: "*"
@@ -69,30 +70,19 @@ class ConfigurationTest < ActiveSupport::TestCase
     configuration = SolidQueue::Configuration.new(dispatchers: [ { polling_interval: 0.1 } ])
 
     assert_processes configuration, :dispatcher, 1, polling_interval: 0.1
+    assert_processes configuration, :scheduler, 1
 
-    dispatcher = configuration.configured_processes.first.instantiate
-    assert_has_recurring_task dispatcher, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
+    scheduler = configuration.configured_processes.second.instantiate
+    assert_has_recurring_task scheduler, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
   end
 
-  test "recurring tasks configuration with no dispatchers uses a default dispatcher" do
+  test "recurring tasks configuration adds a scheduler" do
     configuration = SolidQueue::Configuration.new(dispatchers: [])
 
-    assert_processes configuration, :dispatcher, 1, polling_interval: 1
-
-    dispatcher = configuration.configured_processes.first.instantiate
-    assert_has_recurring_task dispatcher, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
-  end
-
-  test "recurring tasks configuration with multiple dispatchers uses the least busy one" do
-    configuration = SolidQueue::Configuration.new(dispatchers: [ { polling_interval: 0.1 }, { polling_interval: 0.4 }, { polling_interval: 0.2 } ])
-
-    assert_processes configuration, :dispatcher, 3, polling_interval: [ 0.1, 0.2, 0.4 ] # sorted by polling interval
-
-    dispatcher = configuration.configured_processes.last.instantiate
-    assert_has_recurring_task dispatcher, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
+    assert_processes configuration, :scheduler, 1
 
-    dispatchers_without_recurring_tasks = configuration.configured_processes.first(2)
-    assert_nil dispatchers_without_recurring_tasks.map { |d| d.attributes[:recurring_tasks] }.uniq.first
+    scheduler = configuration.configured_processes.first.instantiate
+    assert_has_recurring_task scheduler, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
   end
 
   test "no recurring tasks configuration when explicitly excluded" do
@@ -115,9 +105,9 @@ def assert_processes(configuration, kind, count, **attributes)
       end
     end
 
-    def assert_has_recurring_task(dispatcher, key:, **attributes)
-      assert_equal 1, dispatcher.recurring_schedule.configured_tasks.count
-      task = dispatcher.recurring_schedule.configured_tasks.detect { |t| t.key == key }
+    def assert_has_recurring_task(scheduler, key:, **attributes)
+      assert_equal 1, scheduler.recurring_schedule.configured_tasks.count
+      task = scheduler.recurring_schedule.configured_tasks.detect { |t| t.key == key }
 
       attributes.each do |attr, value|
         assert_equal_value value, task.public_send(attr)
diff --git a/test/unit/dispatcher_test.rb b/test/unit/dispatcher_test.rb
diff --git a/test/unit/log_subscriber_test.rb b/test/unit/log_subscriber_test.rb
diff --git a/test/unit/scheduler_test.rb b/test/unit/scheduler_test.rb
diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb
