Merge branch 'master' into save_safety

Author: cunla

scheduler/redis_models/base.py

@@ -176,7 +176,7 @@ def get(cls, name: str, connection: ConnectionType) -> Optional[Self]:
         try:
             return cls.deserialize(decode_dict(res, set()))
         except Exception as e:
-            logger.warning(f"Failed to deserialize {name}: {e}")
+            logger.warning(f"Failed to deserialize {name}: {e}", exc_info=True)
             return None
 
     @classmethod

scheduler/redis_models/job.py

@@ -122,7 +122,6 @@ def prepare_for_execution(self, worker_name: str, registry: JobNamesRegistry, co
         before execution begins.
         :param worker_name: The name of the worker
         :param registry: The registry to add the job to
-        :param current_pid: The current process id
         :param connection: The connection to the broker
         """
         self.worker_name = worker_name

scheduler/worker/worker.py

@@ -281,7 +281,7 @@ def handle_job_failure(self, job: JobModel, queue: Queue, exc_string: str = "")
         if job.status == JobStatus.FAILED:
             self._model.failed_job_count += 1
             self._model.completed_jobs += 1
-        if job.started_at and job.ended_at:
+        if job.started_at is not None and job.ended_at is not None:
             self._model.total_working_time_ms += (job.ended_at - job.started_at).microseconds / 1000.0
         self._model.save(connection=queue.connection)
 
@@ -613,9 +613,11 @@ def monitor_job_execution_process(self, job: JobModel, queue: Queue) -> None:
                 break
             except JobExecutionMonitorTimeoutException:
                 # job execution process has not exited yet and is still running. Send a heartbeat to keep the worker alive.
-                working_time = (utcnow() - job.started_at).total_seconds()
-                self._model.set_current_job_working_time(working_time, self.connection)
-
+                if job.started_at is not None:
+                    working_time = (utcnow() - job.started_at).total_seconds()
+                    self._model.set_current_job_working_time(working_time, self.connection)
+                else:
+                    logger.warning("[Worker {self.name}/{self._pid}]: job.started_at is None, cannot set working time")
                 # Kill the job from this side if something is really wrong (interpreter lock/etc).
                 if job.timeout != -1 and self._model.current_job_working_time > (job.timeout + 60):
                     self._model.heartbeat(self.connection, self.job_monitoring_interval + 60)
@@ -703,7 +705,7 @@ def execute_in_separate_process(self, job: JobModel, queue: Queue) -> None:
         random.seed()
         self.setup_job_execution_process_signals()
         self._is_job_execution_process = True
-        job = JobModel.get(job.name, self.connection)
+        job = JobModel.get(job.name, queue.connection)
         try:
             self.perform_job(job, queue)
         except:  # noqa