Feature: Dynamic scheduled tasks

Author: cupatea

README.md

@@ -632,6 +632,8 @@ Rails.application.config.after_initialize do # or to_prepare
 end
 ```
 
+You can also dynamically add or remove recurring tasks by creating or deleting SolidQueue::RecurringTask records. It works the same way as with static tasks, except you must set the static field to false. Changes won’t be picked up immediately — they take effect after about a one-minute delay.
+
 It's possible to run multiple schedulers with the same `recurring_tasks` configuration, for example, if you have multiple servers for redundancy, and you run the `scheduler` in more than one of them. To avoid enqueuing duplicate tasks at the same time, an entry in a new `solid_queue_recurring_executions` table is created in the same transaction as the job is enqueued. This table has a unique index on `task_key` and `run_at`, ensuring only one entry per task per time will be created. This only works if you have `preserve_finished_jobs` set to `true` (the default), and the guarantee applies as long as you keep the jobs around.
 
 **Note**: a single recurring schedule is supported, so you can have multiple schedulers using the same schedule, but not multiple schedulers using different configurations.

app/models/solid_queue/recurring_task.rb

@@ -11,6 +11,7 @@ class RecurringTask < Record
     validate :existing_job_class
 
     scope :static, -> { where(static: true) }
+    scope :dynamic, -> { where(static: false) }
 
     has_many :recurring_executions, foreign_key: :task_key, primary_key: :key
 

lib/solid_queue/configuration.rb

@@ -93,7 +93,7 @@ def default_options
       end
 
       def invalid_tasks
-        recurring_tasks.select(&:invalid?)
+        static_recurring_tasks.select(&:invalid?)
       end
 
       def only_work?
@@ -122,8 +122,8 @@ def dispatchers
       end
 
       def schedulers
-        if !skip_recurring_tasks? && recurring_tasks.any?
-          [ Process.new(:scheduler, recurring_tasks: recurring_tasks) ]
+        if !skip_recurring_tasks?
+          [ Process.new(:scheduler, recurring_tasks: static_recurring_tasks) ]
         else
           []
         end
@@ -139,8 +139,8 @@ def dispatchers_options
           .map { |options| options.dup.symbolize_keys }
       end
 
-      def recurring_tasks
-        @recurring_tasks ||= recurring_tasks_config.map do |id, options|
+      def static_recurring_tasks
+        @static_recurring_tasks ||= recurring_tasks_config.map do |id, options|
           RecurringTask.from_configuration(id, **options) if options&.has_key?(:schedule)
         end.compact
       end

lib/solid_queue/scheduler.rb

@@ -30,6 +30,12 @@ def run
         loop do
           break if shutting_down?
 
+          recurring_schedule.update_scheduled_tasks.tap do |updated_tasks|
+            if updated_tasks.any?
+              process.update_columns(metadata: metadata.compact)
+            end
+          end
+
           interruptible_sleep(SLEEP_INTERVAL)
         end
       ensure

lib/solid_queue/scheduler/recurring_schedule.rb

@@ -4,10 +4,11 @@ module SolidQueue
   class Scheduler::RecurringSchedule
     include AppExecutor
 
-    attr_reader :configured_tasks, :scheduled_tasks
+    attr_reader :static_tasks, :configured_tasks, :scheduled_tasks
 
     def initialize(tasks)
-      @configured_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }.select(&:valid?)
+      @static_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }.select(&:valid?)
+      @configured_tasks = @static_tasks + dynamic_tasks
       @scheduled_tasks = Concurrent::Hash.new
     end
 
@@ -17,15 +18,36 @@ def empty?
 
     def schedule_tasks
       wrap_in_app_executor do
-        persist_tasks
-        reload_tasks
+        persist_static_tasks
+        reload_static_tasks
       end
 
       configured_tasks.each do |task|
         schedule_task(task)
       end
     end
 
+    def dynamic_tasks
+      SolidQueue::RecurringTask.dynamic
+    end
+
+    def schedule_new_dynamic_tasks
+      dynamic_tasks.where.not(key: scheduled_tasks.keys).each do |task|
+        schedule_task(task)
+      end
+    end
+
+    def unschedule_old_dynamic_tasks
+      (scheduled_tasks.keys - SolidQueue::RecurringTask.pluck(:key)).each do |key|
+        scheduled_tasks[key].cancel
+        scheduled_tasks.delete(key)
+      end
+    end
+
+    def update_scheduled_tasks
+      schedule_new_dynamic_tasks + unschedule_old_dynamic_tasks
+    end
+
     def schedule_task(task)
       scheduled_tasks[task.key] = schedule(task)
     end
@@ -35,18 +57,22 @@ def unschedule_tasks
       scheduled_tasks.clear
     end
 
+    def static_task_keys
+      static_tasks.map(&:key)
+    end
+
     def task_keys
-      configured_tasks.map(&:key)
+      static_task_keys + dynamic_tasks.map(&:key)
     end
 
     private
-      def persist_tasks
-        SolidQueue::RecurringTask.static.where.not(key: task_keys).delete_all
-        SolidQueue::RecurringTask.create_or_update_all configured_tasks
+      def persist_static_tasks
+        SolidQueue::RecurringTask.static.where.not(key: static_task_keys).delete_all
+        SolidQueue::RecurringTask.create_or_update_all static_tasks
       end
 
-      def reload_tasks
-        @configured_tasks = SolidQueue::RecurringTask.where(key: task_keys)
+      def reload_static_tasks
+        @static_tasks = SolidQueue::RecurringTask.static.where(key: static_task_keys)
       end
 
       def schedule(task)

test/unit/configuration_test.rb

@@ -21,7 +21,7 @@ class ConfigurationTest < ActiveSupport::TestCase
   test "default configuration when config given is empty" do
     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_equal 3, configuration.configured_processes.count # includes scheduler for dynamic tasks
     assert_processes configuration, :worker, 1, queues: "*"
     assert_processes configuration, :dispatcher, 1, batch_size: SolidQueue::Configuration::DISPATCHER_DEFAULTS[:batch_size]
   end
@@ -101,11 +101,11 @@ class ConfigurationTest < ActiveSupport::TestCase
 
     configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_production_only))
     assert configuration.valid?
-    assert_processes configuration, :scheduler, 0
+    assert_processes configuration, :scheduler, 1 # Starts in case of dynamic tasks
 
     configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_empty))
     assert configuration.valid?
-    assert_processes configuration, :scheduler, 0
+    assert_processes configuration, :scheduler, 1 # Starts in case of dynamic tasks
 
     # No processes
     configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [])

test/unit/scheduler_test.rb

@@ -3,7 +3,7 @@
 class SchedulerTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
-  test "recurring schedule" do
+  test "recurring schedule (only static)" do
     recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
     scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks).tap(&:start)
 
@@ -17,6 +17,41 @@ class SchedulerTest < ActiveSupport::TestCase
     scheduler.stop
   end
 
+  test "recurring schedule (only dynamic)" do
+    SolidQueue::RecurringTask.create(
+      key: "dynamic_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
+    )
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks:  {}).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_equal "Scheduler", process.kind
+
+    assert_metadata process, recurring_schedule: [ "dynamic_task" ]
+  ensure
+    scheduler.stop
+  end
+
+  test "recurring schedule (static + dynamic)" do
+    SolidQueue::RecurringTask.create(
+      key: "dynamic_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
+    )
+
+    recurring_tasks = { static_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
+
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_equal "Scheduler", process.kind
+
+    assert_metadata process, recurring_schedule: [ "static_task", "dynamic_task" ]
+  ensure
+    scheduler.stop
+  end
+
   test "run more than one instance of the scheduler with recurring tasks" do
     recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
     schedulers = 2.times.collect do
@@ -33,4 +68,67 @@ class SchedulerTest < ActiveSupport::TestCase
       assert_equal 1, run_at_times[i + 1] - run_at_times[i]
     end
   end
+
+  test "updates metadata after adding dynamic task post-start" do
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}).tap do |s|
+      s.define_singleton_method(:interruptible_sleep) { |interval| sleep 0.1 }
+      s.start
+    end
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    # initially there are no recurring_schedule keys
+    assert process.metadata, {}
+
+    # now create a dynamic task after the scheduler has booted
+    SolidQueue::RecurringTask.create(
+      key:       "new_dynamic_task",
+      static:    false,
+      class_name: "AddToBufferJob",
+      schedule:  "every second",
+      arguments: [ 42 ]
+    )
+
+    sleep 1
+
+    process.reload
+
+    # metadata should now include the new key
+    assert_metadata process, recurring_schedule: [ "new_dynamic_task" ]
+  ensure
+    scheduler&.stop
+  end
+
+  test "updates metadata after removing dynamic task post-start" do
+    old_dynamic_task = SolidQueue::RecurringTask.create(
+      key:       "old_dynamic_task",
+      static:    false,
+      class_name: "AddToBufferJob",
+      schedule:  "every second",
+      arguments: [ 42 ]
+    )
+
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}).tap do |s|
+      s.define_singleton_method(:interruptible_sleep) { |interval| sleep 0.1 }
+      s.start
+    end
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    # initially there is one recurring_schedule key
+    assert_metadata process, recurring_schedule: [ "old_dynamic_task" ]
+
+    old_dynamic_task.destroy
+
+    sleep 1
+
+    process.reload
+
+    # The task is unschedule after it's being removed, and it's reflected in the metadata
+    assert process.metadata, {}
+  ensure
+    scheduler&.stop
+  end
 end