fix:add some logging

Author: cunla

scheduler/helpers/queues/queue_logic.py

@@ -237,9 +237,11 @@ def create_and_enqueue_job(
         )
         if when is None:
             job_model = self.enqueue_job(job_model, connection=pipeline, at_front=at_front)
-        else:
+        elif isinstance(when, datetime):
             job_model.save(connection=self.connection)
-            self.scheduled_job_registry.schedule(self.connection, job_model, when)
+            self.scheduled_job_registry.schedule(self.connection, job_model.name, when)
+        else:
+            raise TypeError(f"Invalid type for when=`{when}`")
         return job_model
 
     def job_handle_success(self, job: JobModel, result: Any, result_ttl: int, connection: ConnectionType):

scheduler/models/task.py

@@ -347,6 +347,7 @@ def _schedule(self) -> bool:
             return False
         schedule_time = self._schedule_time()
         if self.task_type in {TaskType.REPEATABLE, TaskType.ONCE} and schedule_time < timezone.now():
+            logger.debug(f"Task {str(self)} scheduled time is in the past, not scheduling")
             return False
         kwargs = self._enqueue_args()
         job = self.rqueue.create_and_enqueue_job(
@@ -356,7 +357,6 @@ def _schedule(self) -> bool:
             **kwargs,
         )
         self.job_name = job.name
-        super(Task, self).save()
         return True
 
     def save(self, **kwargs):

scheduler/redis_models/registry/queue_registries.py

@@ -60,18 +60,21 @@ def cleanup(self, connection: ConnectionType, timestamp: Optional[float] = None)
         and `all()` methods implemented in JobIdsRegistry."""
         pass
 
-    def schedule(self, connection: ConnectionType, job: JobModel, scheduled_datetime: datetime):
-        """
-        Adds job to registry, scored by its execution time (in UTC).
+    def schedule(self, connection: ConnectionType, job_name: str, scheduled_datetime: datetime) -> int:
+        """Adds job_name to registry, scored by its execution time (in UTC).
         If datetime has no tzinfo, it will assume localtimezone.
+
+        :param connection: Broker connection
+        :param job_name: Job name to schedule
+        :param scheduled_datetime: datetime to schedule job
         """
         # If datetime has no timezone, assume server's local timezone
         if not scheduled_datetime.tzinfo:
             tz = timezone(timedelta(seconds=-(time.timezone if time.daylight == 0 else time.altzone)))
             scheduled_datetime = scheduled_datetime.replace(tzinfo=tz)
 
         timestamp = scheduled_datetime.timestamp()
-        return connection.zadd(self._key, {job.name: timestamp})
+        return self.add(connection=connection, job_name=job_name, score=timestamp)
 
     def get_jobs_to_schedule(self, timestamp: int, chunk_size: int = 1000) -> List[str]:
         """Gets a list of job names that should be scheduled.

scheduler/tests/test_task_types/test_repeatable_task.py

@@ -5,17 +5,31 @@
 from django.utils import timezone
 
 from scheduler import settings
+from scheduler.models import TaskType, Task
 from scheduler.redis_models import JobModel
 from scheduler.tests.test_task_types.test_task_model import BaseTestCases
 from scheduler.tests.testtools import task_factory, _get_task_scheduled_job_from_registry
-from scheduler.models import TaskType
 from scheduler.types import SchedulerConfiguration
 
 
 class TestRepeatableTask(BaseTestCases.TestSchedulableTask):
     task_type = TaskType.REPEATABLE
     queue_name = settings.get_queue_names()[0]
 
+    def test_create_task_error(self):
+        scheduled_time = timezone.now()
+
+        Task.objects.create(
+            name="konichiva_every_2s",
+            callable="chat.task_scheduler.konichiva_func",
+            task_type="REPEATABLE",
+            interval=2,
+            interval_unit="seconds",
+            queue="default",
+            enabled=True,
+            scheduled_time=scheduled_time,
+        )
+
     def test_unschedulable_old_job(self):
         job = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1), repeat=0)
         self.assertFalse(job.is_scheduled())
@@ -145,7 +159,7 @@ def test_repeat_none_interval_2_min(self):
         task = task_factory(self.task_type, scheduled_time=base_time - timedelta(minutes=29), repeat=None)
         task.interval = 120
         task.interval_unit = "seconds"
-        task._schedule()
+        task.save()
         self.assertTrue(task.scheduled_time > base_time)
         self.assertTrue(task.is_scheduled())
 

scheduler/tests/test_task_types/test_task_model.py

@@ -102,7 +102,7 @@ def test_clean_invalid_queue(self):
 
         def test_is_schedulable_already_scheduled(self):
             task = task_factory(self.task_type)
-            task._schedule()
+            task.save()
             self.assertTrue(task.is_scheduled())
 
         def test_is_schedulable_disabled(self):