Track recurring executions to prevent enqueuing the same task more than once

Only when the recurring job being enqueued is using Solid Queue as the adapter.
This supports other adapters as well, but in that case we can't guarantee unique
runs of the same task at the same time.

Author: rosa

app/models/solid_queue/job.rb

@@ -2,7 +2,7 @@
 
 module SolidQueue
   class Job < Record
-    include Executable
+    include Executable, Recurrable, Clearable
 
     serialize :arguments, coder: JSON
 

app/models/solid_queue/job/executable.rb

@@ -6,7 +6,7 @@ module Executable
       extend ActiveSupport::Concern
 
       included do
-        include Clearable, ConcurrencyControls, Schedulable
+        include ConcurrencyControls, Schedulable
 
         has_one :ready_execution
         has_one :claimed_execution

app/models/solid_queue/job/recurrable.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    module Recurrable
+      extend ActiveSupport::Concern
+
+      included do
+        has_one :recurring_execution
+      end
+
+      private
+        def execution
+          super || recurring_execution
+        end
+    end
+  end
+end

app/models/solid_queue/recurring_execution.rb

@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class RecurringExecution < Execution
+    def self.record(task_key, run_at, &block)
+      transaction do
+        if job_id = block.call
+          create!(job_id: job_id, task_key: task_key, run_at: run_at)
+        end
+      end
+    rescue ActiveRecord::RecordNotUnique
+      SolidQueue.logger.info("[SolidQueue] Skipped recurring task #{task_key} at #{run_at} — already dispatched")
+    end
+  end
+end

lib/solid_queue/dispatcher/recurring_schedule.rb

@@ -36,11 +36,11 @@ def inspect
 
     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)
+        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)
 
           wrap_in_app_executor do
-            thread_task.enqueue
+            thread_task.enqueue(at: thread_task_run_at)
           end
         end
 

lib/solid_queue/dispatcher/recurring_task.rb

@@ -29,20 +29,35 @@ def next_time
       schedule.next_time.utc
     end
 
-    def enqueue
-      SolidQueue.logger.info("[SolidQueue] Dispatching recurring task #{self}")
-      job_class.perform_later(*arguments)
+    def enqueue(at:)
+      if using_solid_queue_adapter?
+        perform_later_and_record(run_at: at)
+      else
+        perform_later
+      end
     end
 
     def valid?
       schedule.instance_of?(Fugit::Cron)
     end
 
     def to_s
-      "#{class_name}.perform_later(#{arguments.map(&:inspect).join(",")}) with schedule #{schedule.original}"
+      "#{class_name}.perform_later(#{arguments.map(&:inspect).join(",")}) [ #{schedule.original} ]"
     end
 
     private
+      def using_solid_queue_adapter?
+        job_class.queue_adapter_name.inquiry.solid_queue?
+      end
+
+      def perform_later_and_record(run_at:)
+        RecurringExecution.record(key, run_at) { perform_later.provider_job_id }
+      end
+
+      def perform_later
+        job_class.perform_later(*arguments)
+      end
+
       def job_class
         @job_class ||= class_name.safe_constantize
       end

test/fixtures/solid_queue/recurring_executions.yml

@@ -46,7 +46,7 @@ class DispatcherTest < ActiveSupport::TestCase
 
     process = SolidQueue::Process.first
     assert_equal "Dispatcher", process.kind
-    assert_equal [ "AddToBufferJob.perform_later(42) with schedule 0 * * * *" ], process.metadata["recurring_schedule"]
+    assert_equal [ "AddToBufferJob.perform_later(42) [ 0 * * * * ]" ], process.metadata["recurring_schedule"]
 
     with_recurring_schedule.stop
   end