Watch for changes in dynamic recurring tasks

Author: cupatea

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

@@ -31,6 +31,23 @@ 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

test/unit/scheduler_test.rb

@@ -68,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