Persist and reload recurring tasks before scheduling them

And delete them when the schedule is unloaded.

Author: rosa

app/models/solid_queue/recurring_task.rb

@@ -1,3 +1,5 @@
+# frozen_string_literal: true
+
 require "fugit"
 
 module SolidQueue
@@ -7,6 +9,8 @@ class RecurringTask < Record
     validate :supported_schedule
     validate :existing_job_class
 
+    scope :static, -> { where(static: true) }
+
     class << self
       def wrap(args)
         args.is_a?(self) ? args : from_configuration(args.first, **args.second)
@@ -51,12 +55,8 @@ def to_s
       "#{class_name}.perform_later(#{arguments.map(&:inspect).join(",")}) [ #{parsed_schedule.original} ]"
     end
 
-    def to_h
-      {
-        schedule: schedule,
-        class_name: class_name,
-        arguments: arguments
-      }
+    def attributes_for_upsert
+      attributes.without("id", "created_at", "updated_at")
     end
 
     private
@@ -72,7 +72,6 @@ def existing_job_class
         end
       end
 
-
       def using_solid_queue_adapter?
         job_class.queue_adapter_name.inquiry.solid_queue?
       end

app/models/solid_queue/recurring_task/arguments.rb

@@ -1,3 +1,7 @@
+# frozen_string_literal: true
+
+require "active_job/arguments"
+
 module SolidQueue
   class RecurringTask::Arguments
     class << self
@@ -6,7 +10,7 @@ def load(data)
       end
 
       def dump(data)
-        ActiveSupport::JSON.dump(ActiveJob::Arguments.serialize(data)) unless data.nil?
+        ActiveSupport::JSON.dump(ActiveJob::Arguments.serialize(Array(data)))
       end
     end
   end

lib/solid_queue/dispatcher.rb

@@ -4,8 +4,8 @@ module SolidQueue
   class Dispatcher < Processes::Poller
     attr_accessor :batch_size, :concurrency_maintenance, :recurring_schedule
 
-    after_boot :start_concurrency_maintenance, :load_recurring_schedule
-    before_shutdown :stop_concurrency_maintenance, :unload_recurring_schedule
+    after_boot :start_concurrency_maintenance, :schedule_recurring_tasks
+    before_shutdown :stop_concurrency_maintenance, :unschedule_recurring_tasks
 
     def initialize(**options)
       options = options.dup.with_defaults(SolidQueue::Configuration::DISPATCHER_DEFAULTS)
@@ -19,7 +19,7 @@ def initialize(**options)
     end
 
     def metadata
-      super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval, recurring_schedule: recurring_schedule.tasks.presence)
+      super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval, recurring_schedule: recurring_schedule.task_keys.presence)
     end
 
     private
@@ -38,16 +38,16 @@ def start_concurrency_maintenance
         concurrency_maintenance&.start
       end
 
-      def load_recurring_schedule
-        recurring_schedule.load_tasks
+      def schedule_recurring_tasks
+        recurring_schedule.schedule_tasks
       end
 
       def stop_concurrency_maintenance
         concurrency_maintenance&.stop
       end
 
-      def unload_recurring_schedule
-        recurring_schedule.unload_tasks
+      def unschedule_recurring_tasks
+        recurring_schedule.unschedule_tasks
       end
 
       def all_work_completed?

lib/solid_queue/dispatcher/recurring_schedule.rb

@@ -7,41 +7,56 @@ class Dispatcher::RecurringSchedule
     attr_reader :configured_tasks, :scheduled_tasks
 
     def initialize(tasks)
-      @configured_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }
+      @configured_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }.select(&:valid?)
       @scheduled_tasks = Concurrent::Hash.new
     end
 
     def empty?
       configured_tasks.empty?
     end
 
-    def load_tasks
+    def schedule_tasks
+      wrap_in_app_executor do
+        persist_tasks
+        reload_tasks
+      end
+
       configured_tasks.each do |task|
-        load_task(task)
+        schedule_task(task)
       end
     end
 
-    def load_task(task)
+    def schedule_task(task)
       scheduled_tasks[task.key] = schedule(task)
     end
 
-    def unload_tasks
+    def unschedule_tasks
       scheduled_tasks.values.each(&:cancel)
       scheduled_tasks.clear
-    end
 
-    def tasks
-      configured_tasks.each_with_object({}) { |task, hsh| hsh[task.key] = task.to_h }
+      wrap_in_app_executor { delete_tasks }
     end
 
-    def inspect
-      configured_tasks.map(&:to_s).join(" | ")
+    def task_keys
+      configured_tasks.map(&:key)
     end
 
     private
+      def persist_tasks
+        SolidQueue::RecurringTask.upsert_all configured_tasks.map(&:attributes_for_upsert), record_timestamps: true
+      end
+
+      def reload_tasks
+        @configured_tasks = SolidQueue::RecurringTask.where(key: task_keys)
+      end
+
+      def delete_tasks
+        SolidQueue::RecurringTask.static.delete_all
+      end
+
       def schedule(task)
         scheduled_task = Concurrent::ScheduledTask.new(task.delay_from_now, args: [ self, task, task.next_time ]) do |thread_schedule, thread_task, thread_task_run_at|
-          thread_schedule.load_task(thread_task)
+          thread_schedule.schedule_task(thread_task)
 
           wrap_in_app_executor do
             thread_task.enqueue(at: thread_task_run_at)

test/integration/instrumentation_test.rb

@@ -306,7 +306,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     assert events.size >= 2
     events.each do |event|
-      assert_event event, "enqueue_recurring_task", task: :example_task
+      assert_event event, "enqueue_recurring_task", task: "example_task"
     end
 
     active_job_ids = SolidQueue::Job.all.map(&:active_job_id)
@@ -333,7 +333,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert events.size >= 1
     event = events.last
 
-    assert_event event, "enqueue_recurring_task", task: :example_task, enqueue_error: "ActiveRecord::Deadlocked: ActiveRecord::Deadlocked"
+    assert_event event, "enqueue_recurring_task", task: "example_task", enqueue_error: "ActiveRecord::Deadlocked: ActiveRecord::Deadlocked"
     assert event.last[:at].present?
     assert_nil event.last[:other_adapter]
   end
@@ -354,7 +354,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert events.size >= 1
     event = events.last
 
-    assert_event event, "enqueue_recurring_task", task: :example_task, enqueue_error: "All is broken"
+    assert_event event, "enqueue_recurring_task", task: "example_task", enqueue_error: "All is broken"
     assert event.last[:at].present?
     assert event.last[:other_adapter]
   ensure

test/integration/recurring_tasks_test.rb

@@ -16,6 +16,7 @@ class RecurringTasksTest < ActiveSupport::TestCase
 
     SolidQueue::Process.destroy_all
     SolidQueue::Job.destroy_all
+    SolidQueue::RecurringTask.delete_all
     JobResult.delete_all
   end
 
@@ -40,8 +41,56 @@ class RecurringTasksTest < ActiveSupport::TestCase
     end
   end
 
+  test "persist and delete configured tasks" do
+    configured_task = { periodic_store_result: { class: "StoreResultJob", schedule: "every second" } }
+
+    assert_recurring_tasks configured_task
+    terminate_process(@pid)
+    assert_recurring_tasks []
+
+    SolidQueue::RecurringTask.create!(key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every minute", arguments: [ 42 ])
+
+    @pid = run_supervisor_as_fork
+    wait_for_registered_processes(4, timeout: 3.second)
+
+    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)
+
+    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)
+
+    assert_recurring_tasks configured_task.merge(updated_task)
+
+    terminate_process(@pid)
+    dispatcher1.stop
+    dispatcher2.stop
+
+    assert_recurring_tasks []
+  end
+
   private
     def wait_for_jobs_to_be_enqueued(count, timeout: 1.second)
       wait_while_with_timeout(timeout) { SolidQueue::Job.count < count }
     end
+
+    def assert_recurring_tasks(expected_tasks)
+      skip_active_record_query_cache do
+        assert_equal expected_tasks.count, SolidQueue::RecurringTask.count
+
+        expected_tasks.each do |key, attrs|
+          task = SolidQueue::RecurringTask.find_by(key: key)
+          assert task.present?
+
+          assert_equal(attrs[:schedule], task.schedule) if attrs[:schedule]
+          assert_equal(attrs[:class], task.class_name) if attrs[:class]
+          assert_equal(attrs[:args], task.arguments) if attrs[:args]
+        end
+      end
+    end
 end

test/unit/dispatcher_test.rb

@@ -51,8 +51,7 @@ class DispatcherTest < ActiveSupport::TestCase
     assert_equal "Dispatcher", process.kind
 
     schedule_from_metadata = process.metadata["recurring_schedule"]
-    assert_equal 1, schedule_from_metadata.size
-    assert_equal({ "class_name" => "AddToBufferJob", "schedule" => "every hour", "arguments" => [ 42 ] }, schedule_from_metadata["example_task"])
+    assert_equal [ "example_task" ], schedule_from_metadata
   ensure
     with_recurring_schedule.stop
   end