Fix duplicate recurring enqueues across scheduler instances

Author: knifecake

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## Unreleased
+
+**Fixed:**
+
+- Avoid duplicate recurring-task enqueues when multiple schedulers race on the
+  same `run_at`. We now create recurring execution records atomically and skip
+  already-recorded runs, matching Solid Queue's behavior.
+
 ## v0.2.0 - 2026-04-18
 
 **Fixed:**

steady_queue/models/recurring_execution.py

@@ -1,4 +1,4 @@
-from django.db import models
+from django.db import IntegrityError, models
 from django.tasks import TaskResult
 
 from .execution import Execution, ExecutionQuerySet
@@ -9,9 +9,10 @@ def clearable(self):
         return self.filter(job__isnull=True)
 
     def record(self, task_result: TaskResult, task, run_at):
-        self.update_or_create(
-            task=task, run_at=run_at, defaults={"job_id": task_result.id}
-        )
+        try:
+            self.create(task=task, run_at=run_at, job_id=task_result.id)
+        except IntegrityError as e:
+            raise self.model.AlreadyRecorded from e
 
     def clear_in_batches(self, batch_size=500):
         while True:
@@ -21,6 +22,9 @@ def clear_in_batches(self, batch_size=500):
 
 
 class RecurringExecution(Execution):
+    class AlreadyRecorded(Exception):
+        pass
+
     class Meta:
         verbose_name = "recurring execution"
         verbose_name_plural = "recurring executions"

steady_queue/models/recurring_task.py

@@ -97,7 +97,10 @@ def last_enqueued_time(self) -> datetime:
         return self.recurring_executions.max("run_at")
 
     def enqueue(self, run_at: datetime):
-        return self.enqueue_and_record(run_at)
+        try:
+            return self.enqueue_and_record(run_at)
+        except RecurringExecution.AlreadyRecorded:
+            return False
 
     def enqueue_and_record(self, run_at: datetime):
         with transaction.atomic(using=self._state.db):
@@ -108,6 +111,7 @@ def enqueue_and_record(self, run_at: datetime):
                 queue_name=self.queue_name, priority=self.priority
             ).enqueue(*args, **kwargs)
             RecurringExecution.objects.record(task_result, self, run_at)
+            return task_result
 
     @property
     def previous_time(self) -> datetime:

tests/test_recurring_execution.py

@@ -0,0 +1,36 @@
+from django.test import TestCase
+from django.utils import timezone
+
+from steady_queue.configuration import Configuration
+from steady_queue.models import Job, RecurringExecution, RecurringTask
+from tests.dummy.tasks import dummy_task
+
+
+class RecurringExecutionTestCase(TestCase):
+    def test_enqueue_same_run_at_only_enqueues_once(self):
+        recurring_task = RecurringTask.from_configuration(
+            Configuration.RecurringTask(
+                key="test-recurring-task",
+                class_name="tests.dummy.tasks.dummy_task",
+                schedule="* * * * *",
+                arguments=dummy_task.serialize([], {}),
+                queue_name="default",
+                priority=0,
+            )
+        )
+        recurring_task.save()
+
+        run_at = timezone.now().replace(second=0, microsecond=0)
+
+        first_result = recurring_task.enqueue(run_at=run_at)
+        second_result = recurring_task.enqueue(run_at=run_at)
+
+        self.assertNotEqual(first_result, False)
+        self.assertFalse(second_result)
+
+        self.assertEqual(Job.objects.count(), 1)
+        self.assertEqual(RecurringExecution.objects.count(), 1)
+
+        recurring_execution = RecurringExecution.objects.get()
+        self.assertEqual(recurring_execution.task_id, recurring_task.key)
+        self.assertEqual(recurring_execution.run_at, run_at)