Fix recurring task double-enqueue caused by wall-clock race condition

Author: ghiculescu

app/models/solid_queue/recurring_task.rb

@@ -56,8 +56,9 @@ def create_or_update_all(tasks)
       end
     end
 
-    def delay_from_now
-      [ (next_time - Time.current).to_f, 0.1 ].max
+
+    def next_time_after(time)
+      parsed_schedule.next_time(time).utc
     end
 
     def next_time

lib/solid_queue/scheduler/recurring_schedule.rb

@@ -33,8 +33,8 @@ def schedule_tasks
       end
     end
 
-    def schedule_task(task)
-      scheduled_tasks[task.key] = schedule(task)
+    def schedule_task(task, run_at: task.next_time)
+      scheduled_tasks[task.key] = schedule(task, run_at: run_at)
     end
 
     def unschedule_tasks
@@ -99,9 +99,11 @@ def load_dynamic_tasks
         dynamic_tasks_enabled? ? RecurringTask.dynamic.to_a : []
       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.schedule_task(thread_task)
+      def schedule(task, run_at: task.next_time)
+        delay = [ (run_at - Time.current).to_f, 0.1 ].max
+
+        scheduled_task = Concurrent::ScheduledTask.new(delay, args: [ self, task, run_at ]) do |thread_schedule, thread_task, thread_task_run_at|
+          thread_schedule.schedule_task(thread_task, run_at: thread_task.next_time_after(thread_task_run_at))
 
           wrap_in_app_executor do
             thread_task.enqueue(at: thread_task_run_at)

test/models/solid_queue/recurring_task_test.rb

@@ -203,6 +203,18 @@ def perform
     assert_equal 4, job.priority
   end
 
+  test "next_time_after returns the next occurrence after the given time" do
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every minute")
+
+    # next_time_after a time exactly on the minute boundary should return the following minute
+    time = Time.utc(2026, 3, 12, 1, 28, 0)
+    assert_equal Time.utc(2026, 3, 12, 1, 29, 0), task.next_time_after(time)
+
+    # next_time_after a time just before the boundary should return that boundary
+    time = Time.utc(2026, 3, 12, 1, 27, 59)
+    assert_equal Time.utc(2026, 3, 12, 1, 28, 0), task.next_time_after(time)
+  end
+
   test "task configured with a command" do
     task = recurring_task_with(command: "JobBuffer.add('from_a_command')")
     enqueue_and_assert_performed_with_result(task, "from_a_command")