fix:timeout

Author: cunla

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() + self.job_info_ttl)
+        registry.add(connection, self.name, self.last_heartbeat.timestamp() + self.timeout)
         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/worker/worker.py

@@ -35,7 +35,7 @@
 
 from scheduler.helpers.queues import Queue, perform_job
 from scheduler.helpers.timeouts import JobExecutionMonitorTimeoutException, JobTimeoutException
-from scheduler.helpers.utils import utcnow, current_timestamp
+from scheduler.helpers.utils import utcnow
 
 try:
     from setproctitle import setproctitle as setprocname
@@ -96,16 +96,16 @@ def from_model(cls, model: WorkerModel) -> Self:
         return res
 
     def __init__(
-        self,
-        queues: Iterable[Union[str, Queue]],
-        name: str,
-        maintenance_interval: int = SCHEDULER_CONFIG.DEFAULT_MAINTENANCE_TASK_INTERVAL,
-        job_monitoring_interval: int = SCHEDULER_CONFIG.DEFAULT_JOB_MONITORING_INTERVAL,
-        dequeue_strategy: DequeueStrategy = DequeueStrategy.DEFAULT,
-        fork_job_execution: bool = True,
-        with_scheduler: bool = True,
-        burst: bool = False,
-        model: Optional[WorkerModel] = None,
+            self,
+            queues: Iterable[Union[str, Queue]],
+            name: str,
+            maintenance_interval: int = SCHEDULER_CONFIG.DEFAULT_MAINTENANCE_TASK_INTERVAL,
+            job_monitoring_interval: int = SCHEDULER_CONFIG.DEFAULT_JOB_MONITORING_INTERVAL,
+            dequeue_strategy: DequeueStrategy = DequeueStrategy.DEFAULT,
+            fork_job_execution: bool = True,
+            with_scheduler: bool = True,
+            burst: bool = False,
+            model: Optional[WorkerModel] = None,
     ) -> None:
         self.fork_job_execution = fork_job_execution
         self.job_monitoring_interval: int = job_monitoring_interval
@@ -352,7 +352,7 @@ def run_maintenance_tasks(self) -> None:
             self._model.save(connection=self.connection)
 
     def dequeue_job_and_maintain_ttl(
-        self, timeout: Optional[int], max_idle_time: Optional[int] = None
+            self, timeout: Optional[int], max_idle_time: Optional[int] = None
     ) -> Tuple[Optional[JobModel], Optional[Queue]]:
         """Dequeues a job while maintaining the TTL.
         :param timeout: The timeout for the dequeue operation.
@@ -523,7 +523,7 @@ def reorder_queues(self, reference_queue: Queue) -> None:
             return
         if self._dequeue_strategy == DequeueStrategy.ROUND_ROBIN:
             pos = self._ordered_queues.index(reference_queue)
-            self._ordered_queues = self._ordered_queues[pos + 1 :] + self._ordered_queues[: pos + 1]
+            self._ordered_queues = self._ordered_queues[pos + 1:] + self._ordered_queues[: pos + 1]
             return
         if self._dequeue_strategy == DequeueStrategy.RANDOM:
             shuffle(self._ordered_queues)
@@ -607,7 +607,7 @@ def monitor_job_execution_process(self, job: JobModel, queue: Queue) -> None:
         while True:
             try:
                 with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(
-                    self.job_monitoring_interval, JobExecutionMonitorTimeoutException
+                        self.job_monitoring_interval, JobExecutionMonitorTimeoutException
                 ):
                     retpid, ret_val = self.wait_for_job_execution_process()
                 break
@@ -625,7 +625,7 @@ def monitor_job_execution_process(self, job: JobModel, queue: Queue) -> None:
                     self.wait_for_job_execution_process()
                     break
 
-                self.maintain_heartbeats(job, queue)
+                self._model.heartbeat(self.connection, self.job_monitoring_interval + 60)
 
             except OSError as e:
                 # In case we encountered an OSError due to EINTR (which is
@@ -685,11 +685,6 @@ def execute_job(self, job: JobModel, queue: Queue) -> None:
             self.perform_job(job, queue)
             self._model.set_field("state", WorkerStatus.IDLE, connection=self.connection)
 
-    def maintain_heartbeats(self, job: JobModel, queue: Queue) -> None:
-        """Updates worker and job's last heartbeat field."""
-        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.
         After fork()'ing, assure we are generating random sequences that are different from the worker.
@@ -840,7 +835,7 @@ class RoundRobinWorker(Worker):
 
     def reorder_queues(self, reference_queue: Queue) -> None:
         pos = self._ordered_queues.index(reference_queue)
-        self._ordered_queues = self._ordered_queues[pos + 1 :] + self._ordered_queues[: pos + 1]
+        self._ordered_queues = self._ordered_queues[pos + 1:] + self._ordered_queues[: pos + 1]
 
 
 class RandomWorker(Worker):