Fix the time calculation problem caused by start_time

django_celery_beat/models.py

@@ -390,6 +390,11 @@ def from_schedule(cls, schedule):
         except MultipleObjectsReturned:
             return cls.objects.filter(**spec).first()
 
+    def due_start_time(self, start_time, tz):
+        start_time = start_time.astimezone(tz)
+        start, ends_in, now = self.schedule.remaining_delta(start_time)
+        return start + ends_in
+
 
 class PeriodicTasks(models.Model):
     """Helper table for tracking updates to periodic tasks.
@@ -663,3 +668,9 @@ def scheduler(self):
     @property
     def schedule(self):
         return self.scheduler.schedule
+
+    def due_start_time(self, tz):
+        if self.crontab:
+            return self.crontab.due_start_time(self.start_time, tz)
+        else:
+            return self.start_time

django_celery_beat/schedulers.py

@@ -116,9 +116,11 @@ def is_due(self):
             if now < self.model.start_time:
                 # The datetime is before the start date - don't run.
                 # send a delay to retry on start_time
+                now_tz = now.tzinfo
                 delay = math.ceil(
-                    (self.model.start_time - now).total_seconds()
+                    (self.model.due_start_time(now_tz) - now).total_seconds()
                 )
+
                 return schedules.schedstate(False, delay)
 
         # EXPIRED TASK: Disable task when expired

t/unit/test_schedulers.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timedelta
 from itertools import count
 from time import monotonic
+from zoneinfo import ZoneInfo
 
 import pytest
 from celery.schedules import crontab, schedule, solar
@@ -777,6 +778,122 @@ def test_starttime_trigger(self, monkeypatch):
         assert s._heap[0]
         assert s._heap[0][2].name == m1.name
 
+    def test_crontab_with_start_time_between_now_and_crontab(self, app):
+        now = app.now()
+
+        delay_minutes = 2
+
+        test_start_time = now + timedelta(minutes=delay_minutes)
+
+        crontab_time = test_start_time + timedelta(minutes=delay_minutes)
+
+        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 = 2 * delay_minutes * 60
+
+        assert not is_due
+        assert next_check == pytest.approx(expected_delay, abs=60)
+
+    def test_crontab_with_start_time_after_crontab(self, app):
+        now = app.now()
+
+        delay_minutes = 2
+
+        crontab_time = now + timedelta(minutes=delay_minutes)
+
+        test_start_time = crontab_time + timedelta(minutes=delay_minutes)
+
+        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 + 3600
+
+        assert not is_due
+        assert next_check == pytest.approx(expected_delay, abs=60)
+
+    def test_crontab_with_start_time_different_time_zone(self, app):
+        now = app.now()
+
+        delay_minutes = 2
+
+        test_start_time = now + timedelta(minutes=delay_minutes)
+
+        crontab_time = test_start_time + timedelta(minutes=delay_minutes)
+
+        tz = ZoneInfo('Asia/Shanghai')
+        test_start_time = test_start_time.astimezone(tz)
+
+        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 = 2 * delay_minutes * 60
+
+        assert not is_due
+        assert next_check == pytest.approx(expected_delay, abs=60)
+
+        now = app.now()
+
+        crontab_time = now + timedelta(minutes=delay_minutes)
+
+        test_start_time = crontab_time + timedelta(minutes=delay_minutes)
+
+        tz = ZoneInfo('Asia/Shanghai')
+        test_start_time = test_start_time.astimezone(tz)
+
+        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 + 3600
+
+        assert not is_due
+        assert next_check == pytest.approx(expected_delay, abs=60)
+
+    def test_crontab_with_start_time_tick(self, app):
+        PeriodicTask.objects.all().delete()
+        s = self.Scheduler(app=self.app)
+        assert not s._heap
+
+        m1 = self.create_model_interval(schedule(timedelta(seconds=3)))
+        m1.save()
+
+        now = timezone.now()
+        start_time = now + timedelta(minutes=1)
+        crontab_trigger_time = now + timedelta(minutes=2)
+
+        m2 = self.create_model_crontab(
+            crontab(minute=f'{crontab_trigger_time.minute}'),
+            start_time=start_time)
+        m2.save()
+
+        e2 = EntryTrackSave(m2, app=self.app)
+        is_due, _ = e2.is_due()
+
+        while (not is_due):
+            s.tick()
+            assert s._heap[0][2].name != m2.name
+            is_due, _ = e2.is_due()
+
 
 @pytest.mark.django_db
 class test_models(SchedulerCase):