Fix CrontabSchedule task run before schedule time

django_celery_beat/schedulers.py

@@ -94,13 +94,19 @@ def __init__(self, model, app=None):
 
         if not model.last_run_at:
             model.last_run_at = model.date_changed or self._default_now()
-            # if last_run_at is not set and
-            # model.start_time last_run_at should be in way past.
-            # This will trigger the job to run at start_time
-            # and avoid the heap block.
-            if self.model.start_time:
-                model.last_run_at = model.last_run_at \
-                    - datetime.timedelta(days=365 * 30)
+
+            if model.start_time:
+                if isinstance(model.schedule, schedules.schedule) \
+                        and not isinstance(model.schedule, schedules.crontab):
+                    # if last_run_at is not set and
+                    # model.start_time last_run_at should be in way past.
+                    # This will trigger the job to run at start_time
+                    # and avoid the heap block.
+                    model.last_run_at = model.last_run_at \
+                        - datetime.timedelta(days=365 * 30)
+                else:
+                    # last_run_at should be the time the task started.
+                    model.last_run_at = model.start_time
 
         self.last_run_at = model.last_run_at
 

t/unit/test_schedulers.py

@@ -1035,6 +1035,24 @@ def test_crontab_with_start_time_after_crontab(self, app):
         assert not is_due
         assert next_check == pytest.approx(expected_delay, abs=60)
 
+    def test_crontab_with_start_time_before_crontab(self, app):
+        now = app.now()
+        delay_minutes = 2
+        test_start_time = now - timedelta(minutes=delay_minutes)
+        crontab_time = now + timedelta(minutes=delay_minutes)
+
+        # start_time(now - 2min) < now < crontab_time(now + 2min)
+        task = self.create_model_crontab(
+            crontab(minute=f'{crontab_time.minute}'),
+            start_time=test_start_time)
+
+        entry = EntryTrackSave(task, app=app)
+        is_due, next_check = entry.is_due()
+
+        expected_delay = delay_minutes * 60
+        assert not is_due
+        assert next_check < expected_delay
+
     def test_crontab_with_start_time_different_time_zone(self, app):
         now = app.now()
 
@@ -1083,31 +1101,42 @@ def test_crontab_with_start_time_different_time_zone(self, app):
         assert next_check == pytest.approx(expected_delay, abs=60)
 
     def test_crontab_with_start_time_tick(self, app):
+        # Ensure the heapq does not block by new task with start_time
         PeriodicTask.objects.all().delete()
         s = self.Scheduler(app=self.app)
-        assert not s._heap
+        assert s.get_scheduled_tasks() == []
 
         m1 = self.create_model_interval(schedule(timedelta(seconds=3)))
         m1.save()
+        s.tick()
+        assert len(s.get_scheduled_tasks()) == 2
 
         now = timezone.now()
         start_time = now + timedelta(minutes=1)
         crontab_trigger_time = now + timedelta(minutes=2)
 
+        # now < start_time(now + 1min) < crontab_time(now + 2min)
         m2 = self.create_model_crontab(
             crontab(minute=f'{crontab_trigger_time.minute}'),
             start_time=start_time)
         m2.save()
+        s.tick()
+        assert len(s._heap) == 3
+        assert s._heap[0][2].name == m1.name
 
         e2 = EntryTrackSave(m2, app=self.app)
-        is_due, _ = e2.is_due()
+        is_due, delay = e2.is_due()
+        assert not is_due
+        assert 60 < delay < 120
 
-        max_iterations = 1000
-        iterations = 0
-        while (not is_due and iterations < max_iterations):
+        # tick twice to make sure the heap is not blocked by m2
+        # before it reaches its start_time
+        time.sleep(3)
+        s.tick()
+        assert s._heap[0][2].name == m1.name
+        with patch('time.monotonic', side_effect=lambda: monotonic() + 54):
             s.tick()
-            assert s._heap[0][2].name != m2.name
-            is_due, _ = e2.is_due()
+        assert s._heap[0][2].name == m1.name
 
     @pytest.mark.django_db
     def test_crontab_exclusion_logic_basic(self):