fix:bug job with incorrect score to active registry

Author: cunla

scheduler/helpers/queues/queue_logic.py

@@ -397,10 +397,10 @@ def enqueue_job(
     ) -> JobModel:
         """Enqueues a job for delayed execution without checking dependencies.
 
-        If Queue is instantiated with is_async=False, job is executed immediately.
+        If Queue is instantiated with is_async=False, the job is executed immediately.
         :param job_model: The job redis model
         :param pipeline: The Broker Pipeline
-        :param at_front: Whether to enqueue the job at the front
+        :param at_front: Should the job be enqueued at the front
 
         :returns: The enqueued JobModel
         """

scheduler/redis_models/job.py

@@ -128,16 +128,16 @@ def prepare_for_execution(self, worker_name: str, registry: JobNamesRegistry, co
         self.last_heartbeat = utils.utcnow()
         self.started_at = self.last_heartbeat
         self.status = JobStatus.STARTED
-        registry.add(connection, self.name, self.last_heartbeat.timestamp())
+        registry.add(connection, self.name, self.last_heartbeat.timestamp() + self.job_info_ttl)
         self.save(connection=connection)
 
     def after_execution(
-        self,
-        job_info_ttl: int,
-        status: JobStatus,
-        connection: ConnectionType,
-        prev_registry: Optional[JobNamesRegistry] = None,
-        new_registry: Optional[JobNamesRegistry] = None,
+            self,
+            job_info_ttl: int,
+            status: JobStatus,
+            connection: ConnectionType,
+            prev_registry: Optional[JobNamesRegistry] = None,
+            new_registry: Optional[JobNamesRegistry] = None,
     ) -> None:
         """After the job is executed, update the status, heartbeat, and other metadata."""
         self.status = status
@@ -190,26 +190,26 @@ def deserialize(cls, data: Dict[str, Any]) -> Self:
 
     @classmethod
     def create(
-        cls,
-        connection: ConnectionType,
-        func: FunctionReferenceType,
-        queue_name: str,
-        args: Union[List[Any], Optional[Tuple]] = None,
-        kwargs: Optional[Dict[str, Any]] = None,
-        result_ttl: Optional[int] = None,
-        job_info_ttl: Optional[int] = None,
-        status: Optional[JobStatus] = None,
-        description: Optional[str] = None,
-        timeout: Optional[int] = None,
-        name: Optional[str] = None,
-        task_type: Optional[str] = None,
-        scheduled_task_id: Optional[int] = None,
-        meta: Optional[Dict[str, Any]] = None,
-        *,
-        on_success: Optional[Callback] = None,
-        on_failure: Optional[Callback] = None,
-        on_stopped: Optional[Callback] = None,
-        at_front: Optional[bool] = None,
+            cls,
+            connection: ConnectionType,
+            func: FunctionReferenceType,
+            queue_name: str,
+            args: Union[List[Any], Optional[Tuple]] = None,
+            kwargs: Optional[Dict[str, Any]] = None,
+            result_ttl: Optional[int] = None,
+            job_info_ttl: Optional[int] = None,
+            status: Optional[JobStatus] = None,
+            description: Optional[str] = None,
+            timeout: Optional[int] = None,
+            name: Optional[str] = None,
+            task_type: Optional[str] = None,
+            scheduled_task_id: Optional[int] = None,
+            meta: Optional[Dict[str, Any]] = None,
+            *,
+            on_success: Optional[Callback] = None,
+            on_failure: Optional[Callback] = None,
+            on_stopped: Optional[Callback] = None,
+            at_front: Optional[bool] = None,
     ) -> Self:
         """Creates a new job-model for the given function, arguments, and keyword arguments.
         :returns: A job-model instance.
@@ -283,7 +283,7 @@ def create(
 
 
 def _get_call_string(
-    func_name: Optional[str], args: Any, kwargs: Dict[Any, Any], max_length: Optional[int] = None
+        func_name: Optional[str], args: Any, kwargs: Dict[Any, Any], max_length: Optional[int] = None
 ) -> Optional[str]:
     """
     Returns a string representation of the call, formatted as a regular

scheduler/redis_models/worker.py

@@ -84,11 +84,13 @@ def set_current_job_working_time(self, job_execution_time: float, connection: Co
         self.set_field("current_job_working_time", job_execution_time, connection=connection)
 
     def heartbeat(self, connection: ConnectionType, timeout: Optional[int] = None) -> None:
-        timeout = timeout or DEFAULT_WORKER_TTL + 60
-        connection.expire(self._key, timeout)
-        now = utcnow()
-        self.set_field("last_heartbeat", now, connection=connection)
-        logger.debug(f"Next heartbeat for worker {self._key} should arrive in {timeout} seconds.")
+        with connection.pipeline() as pipeline:
+            timeout = timeout or DEFAULT_WORKER_TTL + 60
+            pipeline.expire(self._key, timeout)
+            now = utcnow()
+            self.set_field("last_heartbeat", now, connection=pipeline)
+            pipeline.execute()
+            logger.debug(f"Next heartbeat for worker {self._key} should arrive in {timeout} seconds.")
 
     @classmethod
     def cleanup(cls, connection: ConnectionType, queue_name: Optional[str] = None):

scheduler/worker/worker.py

@@ -687,14 +687,8 @@ def execute_job(self, job: JobModel, queue: Queue) -> None:
 
     def maintain_heartbeats(self, job: JobModel, queue: Queue) -> None:
         """Updates worker and job's last heartbeat field."""
-        with self.connection.pipeline() as pipeline:
-            self._model.heartbeat(pipeline, self.job_monitoring_interval + 60)
-            ttl = self.get_heartbeat_ttl(job)
-
-            queue.active_job_registry.add(pipeline, job.name, current_timestamp() + ttl, update_existing_only=False)
-            results = pipeline.execute()
-            if results[2] == 1:
-                job.delete(queue.connection)
+        self._model.heartbeat(self.connection, self.job_monitoring_interval + 60)
+        ttl = self.get_heartbeat_ttl(job)
 
     def execute_in_separate_process(self, job: JobModel, queue: Queue) -> None:
         """This is the entry point of the newly spawned job execution process.
@@ -785,10 +779,8 @@ def perform_job(self, job: JobModel, queue: Queue) -> bool:
         logger.debug(f"[Worker {self.name}/{self._pid}]: Performing {job.name} code.")
 
         try:
-            with self.connection.pipeline() as pipeline:
-                self.worker_before_execution(job, connection=pipeline)
-                job.prepare_for_execution(self.name, queue.active_job_registry, connection=pipeline)
-                pipeline.execute()
+            self.worker_before_execution(job, connection=queue.connection)
+            job.prepare_for_execution(self.name, queue.active_job_registry, connection=queue.connection)
             timeout = job.timeout or SCHEDULER_CONFIG.DEFAULT_JOB_TIMEOUT
             with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(timeout, JobTimeoutException, job_name=job.name):
                 logger.debug(f"[Worker {self.name}/{self._pid}]: Performing job `{job.name}`...")