fix:infinite loop on callback calling is_scheduled()

Author: cunla

poetry.lock

@@ -57,6 +57,7 @@ coverage = "^7"
 fakeredis = { version = "^2.19", extras = ['lua'] }
 Flake8-pyproject = "^1.2"
 pyyaml = "^6.0"
+freezegun = "^1.2.2"
 
 [tool.poetry.extras]
 yaml = ["pyyaml"]

pyproject.toml

@@ -33,9 +33,10 @@ def create_job_from_dict(job_dict: Dict[str, Any], update):
     del kwargs['callable_args']
     del kwargs['callable_kwargs']
     if kwargs.get('scheduled_time', None):
-        kwargs['scheduled_time'] = timezone.datetime.fromisoformat(kwargs['scheduled_time'])
-        if not settings.USE_TZ:
-            kwargs['scheduled_time'] = timezone.make_naive(kwargs['scheduled_time'])
+        target = timezone.datetime.fromisoformat(kwargs['scheduled_time'])
+        if not settings.USE_TZ and not timezone.is_naive(target):
+            target = timezone.make_naive(target)
+        kwargs['scheduled_time'] = target
     model_fields = set(map(lambda field: field.attname, model._meta.get_fields()))
     keys_to_ignore = list(filter(lambda k: k not in model_fields, kwargs.keys()))
     for k in keys_to_ignore:

scheduler/management/commands/import.py

@@ -5,9 +5,11 @@
 
 import croniter
 from django.apps import apps
+from django.conf import settings as django_settings
 from django.contrib import admin
 from django.contrib.contenttypes.fields import GenericRelation
 from django.core.exceptions import ValidationError
+from django.core.mail import mail_admins
 from django.db import models
 from django.templatetags.tz import utc
 from django.urls import reverse
@@ -24,14 +26,27 @@
 SCHEDULER_INTERVAL = settings.SCHEDULER_CONFIG['SCHEDULER_INTERVAL']
 
 
-def callback_save_job(job, connection, result, *args, **kwargs):
+def failure_callback(job, connection, result, *args, **kwargs):
+    model_name = job.meta.get('task_type', None)
+    scheduled_task_id = job.meta.get('scheduled_task_id', None)
+    if model_name is None or scheduled_task_id:
+        return
+    model = apps.get_model(app_label='scheduler', model_name=model_name)
+    task = model.objects.filter(id=scheduled_task_id).first()
+    mail_admins(f'Task {task.id}/{task.name} has failed',
+                'See django-admin for logs', )
+    pass
+
+
+def success_callback(job, connection, result, *args, **kwargs):
     model_name = job.meta.get('task_type', None)
     if model_name is None:
         return
     model = apps.get_model(app_label='scheduler', model_name=model_name)
     task = model.objects.filter(job_id=job.id).first()
-    if task is not None:
-        task.force_schedule()
+    if task is None:
+        return
+    task.schedule()
 
 
 class BaseTask(models.Model):
@@ -107,12 +122,12 @@ def function_string(self) -> str:
         return self.callable + f"({', '.join(args_list + kwargs_list)})"
 
     def parse_args(self):
-        """Parse args for running job"""
+        """Parse args for running the job"""
         args = self.callable_args.all()
         return [arg.value() for arg in args]
 
     def parse_kwargs(self):
-        """Parse kwargs for running job"""
+        """Parse kwargs for running the job"""
         kwargs = self.callable_kwargs.all()
         return dict([kwarg.value() for kwarg in kwargs])
 
@@ -122,12 +137,12 @@ def _next_job_id(self):
         return f'{self.queue}:{name}:{addition}'
 
     def _enqueue_args(self) -> Dict:
-        """args for DjangoQueue.enqueue.
+        """Args for DjangoQueue.enqueue.
         Set all arguments for DjangoQueue.enqueue/enqueue_at.
         Particularly:
         - set job timeout and ttl
-        - ensure a callback to reschedule job next iteration.
-        - set job-id to proper format
+        - ensure a callback to reschedule the job next iteration.
+        - Set job-id to proper format
         - set job meta
         """
         res = dict(
@@ -136,8 +151,8 @@ def _enqueue_args(self) -> Dict:
                 task_type=self.TASK_TYPE,
                 scheduled_task_id=self.id,
             ),
-            on_success=callback_save_job,
-            on_failure=callback_save_job,
+            on_success=success_callback,
+            on_failure=failure_callback,
             job_id=self._next_job_id(),
         )
         if self.at_front:
@@ -150,37 +165,31 @@ def _enqueue_args(self) -> Dict:
 
     @property
     def rqueue(self) -> DjangoQueue:
-        """Returns django-queue for job
-        """
+        """Returns redis-queue for job"""
         return get_queue(self.queue)
 
     def ready_for_schedule(self) -> bool:
-        """Is task ready to be scheduled?
+        """Is the task ready to be scheduled?
 
-        If task is already scheduled or disabled, then it is not
+        If the task is already scheduled or disabled, then it is not
         ready to be scheduled.
 
-        :returns: True if task is ready to be scheduled.
+        :returns: True if the task is ready to be scheduled.
         """
         if self.is_scheduled():
-            logger.debug(f'Job {self.name} already scheduled')
+            logger.debug(f'Task {self.name} already scheduled')
             return False
         if not self.enabled:
-            logger.debug(f'Job {str(self)} disabled, enable job before scheduling')
+            logger.debug(f'Task {str(self)} disabled, enable task before scheduling')
             return False
         return True
 
     def schedule(self) -> bool:
-        """Schedule job to run.
-        :returns: True if job was scheduled, False otherwise.
+        """Schedule the next execution for the task to run.
+        :returns: True if a job was scheduled, False otherwise.
         """
         if not self.ready_for_schedule():
             return False
-        self.force_schedule()
-        return True
-
-    def force_schedule(self):
-        """Schedule task regardless of its current status"""
         schedule_time = self._schedule_time()
         kwargs = self._enqueue_args()
         job = self.rqueue.enqueue_at(
@@ -190,6 +199,7 @@ def force_schedule(self):
             **kwargs, )
         self.job_id = job.id
         super(BaseTask, self).save()
+        return True
 
     def enqueue_to_run(self) -> bool:
         """Enqueue job to run now."""
@@ -218,7 +228,7 @@ def unschedule(self) -> bool:
         return True
 
     def _schedule_time(self):
-        return utc(self.scheduled_time)
+        return utc(self.scheduled_time) if django_settings.USE_TZ else self.scheduled_time
 
     def to_dict(self) -> Dict:
         """Export model to dictionary, so it can be saved as external file backup"""
@@ -258,10 +268,10 @@ def save(self, **kwargs):
         update_fields = kwargs.get('update_fields', None)
         if update_fields:
             kwargs['update_fields'] = set(update_fields).union({'modified'})
-        super(BaseTask, self).save(**kwargs)
         if schedule_job:
-            self.schedule()
-            super(BaseTask, self).save()
+            self.schedule()  # schedule() already calls save()
+        else:
+            super(BaseTask, self).save(**kwargs)
 
     def delete(self, **kwargs):
         self.unschedule()
@@ -371,16 +381,20 @@ def _enqueue_args(self):
         res['meta']['interval'] = self.interval_seconds()
         return res
 
+    def _schedule_time(self):
+        _now = timezone.now()
+        if self.scheduled_time >= _now:
+            return super()._schedule_time()
+        gap = math.ceil((_now.timestamp() - self.scheduled_time.timestamp()) / self.interval_seconds())
+        if self.repeat is None or self.repeat >= gap:
+            self.scheduled_time += timedelta(seconds=self.interval_seconds() * gap)
+            self.repeat = (self.repeat - gap) if self.repeat is not None else None
+        return super()._schedule_time()
+
     def ready_for_schedule(self):
         if super(RepeatableTask, self).ready_for_schedule() is False:
             return False
-        if self.scheduled_time < timezone.now():
-            gap = math.ceil((timezone.now().timestamp() - self.scheduled_time.timestamp()) / self.interval_seconds())
-            if self.repeat is None or self.repeat >= gap:
-                self.scheduled_time += timedelta(seconds=self.interval_seconds() * gap)
-                self.repeat = (self.repeat - gap) if self.repeat is not None else None
-
-        if self.scheduled_time < timezone.now():
+        if self._schedule_time() < timezone.now():
             return False
         return True
 
@@ -411,7 +425,8 @@ def clean_cron_string(self):
             raise ValidationError({'cron_string': ValidationError(_(str(e)), code='invalid')})
 
     def _schedule_time(self):
-        return tools.get_next_cron_time(self.cron_string)
+        self.scheduled_time = tools.get_next_cron_time(self.cron_string)
+        return super()._schedule_time()
 
     class Meta:
         verbose_name = _('Cron Task')

scheduler/models/scheduled_task.py

@@ -6,6 +6,7 @@
 from django.test import override_settings
 from django.urls import reverse
 from django.utils import timezone
+from freezegun import freeze_time
 
 from scheduler import settings
 from scheduler.models import BaseTask, CronTask, TaskArg, TaskKwarg, RepeatableTask, ScheduledTask
@@ -440,14 +441,20 @@ class TestSchedulableJob(TestBaseJob):
         # Currently ScheduledJob and RepeatableJob
         TaskModelClass = ScheduledTask
 
-        def test_schedule_time_utc(self):
+        @freeze_time("2016-12-25")
+        @override_settings(USE_TZ=False)
+        def test_schedule_time_no_tz(self):
+            task = task_factory(self.TaskModelClass)
+            task.scheduled_time = datetime(2016, 12, 25, 8, 0, 0, tzinfo=None)
+            self.assertEqual("2016-12-25T08:00:00", task._schedule_time().isoformat())
+
+        @freeze_time("2016-12-25")
+        @override_settings(USE_TZ=True)
+        def test_schedule_time_with_tz(self):
             task = task_factory(self.TaskModelClass)
             est = zoneinfo.ZoneInfo('US/Eastern')
-            scheduled_time = datetime(2016, 12, 25, 8, 0, 0, tzinfo=est)
-            task.scheduled_time = scheduled_time
-            utc = zoneinfo.ZoneInfo('UTC')
-            expected = scheduled_time.astimezone(utc).isoformat()
-            self.assertEqual(expected, task._schedule_time().isoformat())
+            task.scheduled_time = datetime(2016, 12, 25, 8, 0, 0, tzinfo=est)
+            self.assertEqual("2016-12-25T13:00:00+00:00", task._schedule_time().isoformat())
 
         def test_result_ttl_passthrough(self):
             job = task_factory(self.TaskModelClass, result_ttl=500)

scheduler/tests/test_models.py

@@ -3,7 +3,6 @@
 
 import croniter
 from django.apps import apps
-from django.templatetags.tz import utc
 from django.utils import timezone
 
 from scheduler.queues import get_queues, logger, get_queue
@@ -25,8 +24,8 @@ def get_next_cron_time(cron_string) -> timezone.datetime:
     with a cron string"""
     now = timezone.now()
     itr = croniter.croniter(cron_string, now)
-    return utc(itr.get_next(timezone.datetime))
-
+    next_itr = itr.get_next(timezone.datetime)
+    return next_itr
 
 def get_scheduled_task(task_model: str, task_id: int):
     if task_model not in MODEL_NAMES: