fix:add some logging

Author: cunla

scheduler/helpers/queues/queue_logic.py

@@ -244,10 +244,12 @@ def create_and_enqueue_job(
             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):
+    def job_handle_success(
+        self, job: JobModel, result: Any, job_info_ttl: int, result_ttl: int, connection: ConnectionType
+    ):
         """Saves and cleanup job after successful execution"""
         job.after_execution(
-            result_ttl,
+            job_info_ttl,
             JobStatus.FINISHED,
             prev_registry=self.active_job_registry,
             new_registry=self.finished_job_registry,
@@ -280,40 +282,26 @@ def job_handle_failure(self, status: JobStatus, job: JobModel, exc_string: str,
             exc_string=exc_string,
         )
 
-    def run_job(self, job: JobModel) -> JobModel:
-        """Run the job
-        :param job: The job to run
-        :returns: The job result
-        """
+    def run_sync(self, job: JobModel) -> JobModel:
+        """Run a job synchronously, meaning on the same process the method was called."""
+        job.prepare_for_execution("sync", self.active_job_registry, self.connection)
         try:
             result = perform_job(job, self.connection)
 
-            result_ttl = job.success_ttl
             with self.connection.pipeline() as pipeline:
-                self.job_handle_success(job, result=result, result_ttl=result_ttl, connection=pipeline)
-                job.expire(result_ttl, connection=pipeline)
+                self.job_handle_success(
+                    job, result=result, job_info_ttl=job.job_info_ttl, result_ttl=job.success_ttl, connection=pipeline
+                )
+
                 pipeline.execute()
-        except Exception as e:
+        except Exception as e:  # noqa
             logger.warning(f"Job {job.name} failed with exception: {e}")
             with self.connection.pipeline() as pipeline:
                 exc_string = "".join(traceback.format_exception(*sys.exc_info()))
                 self.job_handle_failure(JobStatus.FAILED, job, exc_string, pipeline)
                 pipeline.execute()
         return job
 
-    def run_sync(self, job: JobModel) -> JobModel:
-        """Run a job synchronously, meaning on the same process the method was called."""
-        job.prepare_for_execution("sync", self.active_job_registry, self.connection)
-
-        try:
-            self.run_job(job)
-        except:  # noqa
-            with self.connection.pipeline() as pipeline:
-                exc_string = "".join(traceback.format_exception(*sys.exc_info()))
-                self.job_handle_failure(JobStatus.FAILED, job, exc_string, pipeline)
-                pipeline.execute()
-        return job
-
     @classmethod
     def dequeue_any(
         cls,

scheduler/redis_models/job.py

@@ -126,7 +126,7 @@ def prepare_for_execution(self, worker_name: str, registry: JobNamesRegistry, co
 
     def after_execution(
         self,
-        result_ttl: int,
+        job_info_ttl: int,
         status: JobStatus,
         connection: ConnectionType,
         prev_registry: Optional[JobNamesRegistry] = None,
@@ -138,8 +138,8 @@ def after_execution(
         self.last_heartbeat = self.ended_at
         if prev_registry is not None:
             prev_registry.delete(connection, self.name)
-        if new_registry is not None and result_ttl != 0:
-            new_registry.add(connection, self.name, current_timestamp() + result_ttl)
+        if new_registry is not None and job_info_ttl != 0:
+            new_registry.add(connection, self.name, current_timestamp() + job_info_ttl)
         self.save(connection=connection)
 
     @property

scheduler/tests/test_views/test_job_detail_action.py

@@ -88,9 +88,7 @@ def test_single_job_action_enqueue_job_sync_queue(self):
             job = queue.create_and_enqueue_job(test_job)
             job_list.append(job)
 
-        # This job is deferred
-
-        self.assertEqual(job_list[-1].get_status(connection=queue.connection), JobStatus.FAILED)
+        self.assertEqual(job_list[-1].status, JobStatus.FINISHED)
         self.assertIsNotNone(job_list[-1].enqueued_at)
 
         # Try to force enqueue last job should do nothing
@@ -102,3 +100,4 @@ def test_single_job_action_enqueue_job_sync_queue(self):
         tmp = JobModel.get(job_list[-1].name, connection=queue.connection)
         self.assertEqual(tmp.get_status(connection=queue.connection), JobStatus.FINISHED)
         self.assertIsNotNone(tmp.enqueued_at)
+        self.assertGreater(tmp.enqueued_at, job_list[-1].enqueued_at)

scheduler/worker/worker.py

@@ -792,7 +792,13 @@ def handle_job_success(self, job: JobModel, return_value: Any, queue: Queue):
         with self.connection.pipeline() as pipeline:
             while True:
                 try:
-                    queue.job_handle_success(job, result=return_value, result_ttl=job.success_ttl, connection=pipeline)
+                    queue.job_handle_success(
+                        job,
+                        result=return_value,
+                        job_info_ttl=job.job_info_ttl,
+                        result_ttl=job.success_ttl,
+                        connection=pipeline,
+                    )
                     self._model.current_job_name = None
                     self._model.successful_job_count += 1
                     self._model.completed_jobs += 1
@@ -801,7 +807,6 @@ def handle_job_success(self, job: JobModel, return_value: Any, queue: Queue):
 
                     job.expire(job.success_ttl, connection=pipeline)
                     logger.debug(f"[Worker {self.name}/{self._pid}]: Removing job {job.name} from active_job_registry")
-                    queue.active_job_registry.delete(pipeline, job.name)
                     pipeline.execute()
                     logger.debug(
                         f"[Worker {self.name}/{self._pid}]: Finished handling successful execution of job {job.name}"