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

  When a recurring task's ScheduledTask fires, it computes the next
  occurrence by calling `task.next_time`, which uses `Time.current` as the
  reference point. If the timer fires slightly early (e.g. 2ms before the
  minute boundary), or if there's any sub-second timing drift, `next_time`
  can return the *current* slot instead of the next one — causing a
  duplicate enqueue and a near-immediate re-fire.

  Example with "every minute" schedule:

    ScheduledTask fires at 11:59:59.998
    1. Enqueues job for 12:00:00 ✓
    2. Calls next_time → Time.current is 11:59:59.998
       → fugit returns 12:00:00 (the slot we just enqueued)
       → Schedules new ScheduledTask with ~2ms delay
    3. New task fires, enqueues duplicate job for 12:00:00 ✗

  Fix: compute the next occurrence from the known run_at of the slot that
  just fired, rather than from wall clock time. `next_time_after(run_at)`
  passes the time to fugit's `next_time(from)` parameter, so
  `next_time_after(12:00:00)` always returns 12:01:00 regardless of what
  the wall clock says.

Author: ghiculescu

app/models/solid_queue/recurring_task.rb

@@ -51,6 +51,10 @@ def delay_from_now
       [ (next_time - Time.current).to_f, 0.1 ].max
     end
 
+    def next_time_after(time)
+      parsed_schedule.next_time(time).utc
+    end
+
     def next_time
       parsed_schedule.next_time.utc
     end

lib/solid_queue/scheduler/recurring_schedule.rb

@@ -26,8 +26,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
@@ -49,9 +49,11 @@ def reload_tasks
         @configured_tasks = SolidQueue::RecurringTask.where(key: task_keys).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")

test/unit/scheduler_test.rb

@@ -17,6 +17,28 @@ class SchedulerTest < ActiveSupport::TestCase
     scheduler.stop
   end
 
+  test "single scheduler does not double-enqueue recurring tasks" do
+    recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks)
+    scheduler.start
+
+    wait_while_with_timeout(4.seconds) { SolidQueue::RecurringExecution.count < 3 }
+    scheduler.stop
+
+    skip_active_record_query_cache do
+      # Each run_at should appear exactly once — no duplicates
+      run_at_counts = SolidQueue::RecurringExecution.group(:run_at).count
+      duplicates = run_at_counts.select { |_, count| count > 1 }
+      assert_empty duplicates, "Expected no duplicate run_at values, but found: #{duplicates.inspect}"
+
+      # Number of jobs should match number of recurring executions (no extra jobs)
+      assert_equal SolidQueue::Job.count, SolidQueue::RecurringExecution.count,
+        "Expected one job per recurring execution, got #{SolidQueue::Job.count} jobs for #{SolidQueue::RecurringExecution.count} executions"
+    end
+  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