Implement basic recurring task parsing and loading in a schedule

Using concurrent-ruby's scheduled tasks. Each task schedules the next one,
like GoodJob does. Add a simple test and allow dispatcher to be initialized
without having to pass instantiated recurring tasks.

Author: rosa

lib/solid_queue/configuration.rb

@@ -35,7 +35,7 @@ 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.collect { SolidQueue::Worker.new(**worker_options.with_defaults(WORKER_DEFAULTS)) }
+          processes.times.map { Worker.new(**worker_options.with_defaults(WORKER_DEFAULTS)) }
         end
       else
         []
@@ -44,8 +44,10 @@ def workers
 
     def dispatchers
       if mode.in? %i[ dispatch all]
-        dispatchers_options.flat_map do |dispatcher_options|
-          SolidQueue::Dispatcher.new(**dispatcher_options.with_defaults(DISPATCHER_DEFAULTS))
+        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
     end
@@ -75,6 +77,11 @@ def dispatchers_options
           .map { |options| options.dup.symbolize_keys }
       end
 
+      def parse_recurring_tasks(tasks)
+        Array(tasks).map do |id, options|
+          Dispatcher::RecurringTask.from_configuration(id, **options)
+        end.select(&:valid?)
+      end
 
       def load_config_from(file_or_hash)
         case file_or_hash

lib/solid_queue/dispatcher.rb

@@ -4,10 +4,10 @@ module SolidQueue
   class Dispatcher < Processes::Base
     include Processes::Runnable, Processes::Poller
 
-    attr_accessor :batch_size, :concurrency_maintenance, :recurring_tasks
+    attr_accessor :batch_size, :concurrency_maintenance, :recurring_schedule
 
-    after_boot :start_concurrency_maintenance, :schedule_recurring_tasks
-    before_shutdown :stop_concurrency_maintenance, :unschedule_recurring_tasks
+    after_boot :start_concurrency_maintenance, :load_recurring_schedule
+    before_shutdown :stop_concurrency_maintenance, :unload_recurring_schedule
 
     def initialize(**options)
       options = options.dup.with_defaults(SolidQueue::Configuration::DISPATCHER_DEFAULTS)
@@ -16,7 +16,7 @@ def initialize(**options)
       @polling_interval = options[:polling_interval]
 
       @concurrency_maintenance = ConcurrencyMaintenance.new(options[:concurrency_maintenance_interval], options[:batch_size]) if options[:concurrency_maintenance]
-      @recurring_tasks = RecurringTasks.new(options[:recurring_tasks])
+      @recurring_schedule = RecurringSchedule.new(options[:recurring_tasks])
     end
 
     private
@@ -31,28 +31,28 @@ def run
 
       def dispatch_next_batch
         with_polling_volume do
-          SolidQueue::ScheduledExecution.dispatch_next_batch(batch_size)
+          ScheduledExecution.dispatch_next_batch(batch_size)
         end
       end
 
       def start_concurrency_maintenance
         concurrency_maintenance&.start
       end
 
-      def schedule_recurring_tasks
-        recurring_tasks.schedule
+      def load_recurring_schedule
+        recurring_schedule.load_tasks
       end
 
       def stop_concurrency_maintenance
         concurrency_maintenance&.stop
       end
 
-      def unschedule_recurring_tasks
-        recurring_tasks.unschedule
+      def unload_recurring_schedule
+        recurring_schedule.unload_tasks
       end
 
       def metadata
-        super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval)
+        super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval, recurring_schedule: recurring_schedule.tasks.presence)
       end
   end
 end

lib/solid_queue/dispatcher/recurring_schedule.rb

@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Dispatcher::RecurringSchedule
+    include AppExecutor
+
+    attr_reader :configured_tasks, :scheduled_tasks
+
+    def initialize(tasks)
+      @configured_tasks = Array(tasks).map { |task| Dispatcher::RecurringTask.wrap(task) }
+      @scheduled_tasks = Concurrent::Hash.new
+    end
+
+    def load_tasks
+      configured_tasks.each do |task|
+        load_task(task)
+      end
+    end
+
+    def load_task(task)
+      scheduled_tasks[task.id] = schedule(task)
+    end
+
+    def unload_tasks
+      scheduled_tasks.values.each(&:cancel)
+      scheduled_tasks.clear
+    end
+
+    def tasks
+      configured_tasks.map(&:to_s)
+    end
+
+    def inspect
+      tasks.map(&:to_s).join(" | ")
+    end
+
+    private
+      def schedule(task)
+        scheduled_task = Concurrent::ScheduledTask.new(task.delay_from_now, args: [ self, task ]) do |thread_schedule, thread_task|
+          thread_schedule.load_task(task)
+
+          wrap_in_app_executor do
+            thread_task.enqueue
+          end
+        end
+
+        scheduled_task.add_observer do |_, _, error|
+          # Don't notify on task cancellation before execution, as this will happen normally
+          # as part of unloading tasks
+          handle_thread_error(error) if error && !error.is_a?(Concurrent::CancelledOperationError)
+        end
+
+        scheduled_task.tap(&:execute)
+      end
+  end
+end

lib/solid_queue/dispatcher/recurring_task.rb

@@ -0,0 +1,50 @@
+require "fugit"
+
+module SolidQueue
+  class Dispatcher::RecurringTask
+    class << self
+      def wrap(args)
+        args.first.is_a?(self) ? args.first : from_configuration(args.first, **args.second)
+      end
+
+      def from_configuration(id, **options)
+        new(id, class_name: options[:class], schedule: options[:schedule], arguments: options[:args])
+      end
+    end
+
+    attr_reader :id, :schedule, :class_name, :arguments
+
+    def initialize(id, class_name:, schedule:, arguments: nil)
+      @id = id
+      @class_name = class_name
+      @schedule = Fugit.parse(schedule)
+      @arguments = Array(arguments)
+    end
+
+    def delay_from_now
+      [ (next_time - Time.current).to_f, 0 ].max
+    end
+
+    def next_time
+      schedule.next_time.utc
+    end
+
+    def enqueue
+      SolidQueue.logger.info("[SolidQueue] Dispatching recurring task #{self}")
+      job_class.perform_later(*arguments)
+    end
+
+    def valid?
+      schedule.instance_of?(Fugit::Cron)
+    end
+
+    def to_s
+      "#{class_name}.perform_later(#{arguments.map(&:inspect).join(",")}) with schedule #{schedule.original}"
+    end
+
+    private
+      def job_class
+        @job_class ||= class_name.safe_constantize
+      end
+  end
+end

lib/solid_queue/dispatcher/recurring_tasks.rb

@@ -49,7 +49,7 @@
   # config.action_view.annotate_rendered_view_with_filenames = true
 
   logger = ActiveSupport::Logger.new(STDOUT)
-  config.solid_queue.on_thread_error = ->(exception) { logger.error("#{exception.class.name}: #{exception.message}\n#{exception.backtrace.join("\n")}") }
+  config.solid_queue.on_thread_error = ->(exception) { logger.error("#{exception.class.name}: #{exception.message}\n#{(exception.backtrace || caller)&.join("\n")}") }
   config.solid_queue.logger = ActiveSupport::Logger.new(nil)
 
   config.solid_queue.shutdown_timeout = 2.seconds

test/dummy/config/environments/test.rb

@@ -7,6 +7,11 @@ default: &default
   dispatchers:
     - polling_interval: 1
       batch_size: 500
+      recurring_tasks:
+        periodic-add-to-buffer:
+          class: AddToBufferJob
+          args: 42
+          schedule: every second
 
 development:
   <<: *default

test/dummy/config/solid_queue.yml

@@ -32,6 +32,23 @@ class DispatcherTest < ActiveSupport::TestCase
     process = SolidQueue::Process.first
     assert_equal "Dispatcher", process.kind
     assert_equal({ "polling_interval" => 0.1, "batch_size" => 10 }, process.metadata)
+
+    no_concurrency_maintenance_dispatcher.stop
+  end
+
+  test "recurring schedule" do
+    recurring_task = { example_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
+    with_recurring_schedule = SolidQueue::Dispatcher.new(concurrency_maintenance: false, recurring_tasks: recurring_task)
+
+    with_recurring_schedule.start
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_equal "Dispatcher", process.kind
+    assert_equal [ "AddToBufferJob.perform_later(42) with schedule 0 * * * *" ], process.metadata["recurring_schedule"]
+
+    with_recurring_schedule.stop
   end
 
   test "polling queries are logged" do