fix:bug worker added to registry

Author: cunla

scheduler/redis_models/base.py

@@ -76,8 +76,8 @@ def _deserialize(value: str, _type: Type) -> Any:
 class BaseModel:
     name: str
     _element_key_template: ClassVar[str] = ":element:{}"
-    # fields that are not serializable using method above and should be dealt with in the subclass
-    # e.g. args/kwargs for a job
+    # fields that are not serializable using the method above and should be dealt with in the subclass
+    # e.g., args/kwargs for a job
     _non_serializable_fields: ClassVar[Set[str]] = set()
 
     @classmethod

scheduler/redis_models/registry/base_registry.py

@@ -15,7 +15,7 @@ class DequeueTimeout(Exception):
 @dataclasses.dataclass(slots=True, kw_only=True)
 class ZSetModel(BaseModel):
     def cleanup(self, connection: ConnectionType, timestamp: Optional[float] = None) -> None:
-        """Remove expired jobs from registry."""
+        """Remove expired jobs from the registry."""
         score = timestamp or current_timestamp()
         connection.zremrangebyscore(self._key, 0, score)
 
@@ -45,23 +45,23 @@ def __contains__(self, item: str) -> bool:
         return self.connection.zrank(self._key, item) is not None
 
     def all(self, start: int = 0, end: int = -1) -> List[str]:
-        """Returns list of all job names.
+        """Returns a list of all job names.
 
         :param start: Start score/timestamp, default to 0.
         :param end: End score/timestamp, default to -1 (i.e., no max score).
-        :returns: Returns list of all job names with timestamp from start to end
+        :returns: Returns a list of all job names with timestamp from start to end
         """
         self.cleanup(self.connection)
         res = [as_str(job_name) for job_name in self.connection.zrange(self._key, start, end)]
         logger.debug(f"Getting jobs for registry {self._key}: {len(res)} found.")
         return res
 
     def all_with_timestamps(self, start: int = 0, end: int = -1) -> List[Tuple[str, float]]:
-        """Returns list of all job names with their timestamps.
+        """Returns a list of all job names with their timestamps.
 
         :param start: Start score/timestamp, default to 0.
         :param end: End score/timestamp, default to -1 (i.e., no max score).
-        :returns: Returns list of all job names with timestamp from start to end
+        :returns: Returns a list of all job names with timestamp from start to end
         """
         self.cleanup(self.connection)
         res = self.connection.zrange(self._key, start, end, withscores=True)

scheduler/worker/worker.py

@@ -691,10 +691,10 @@ def maintain_heartbeats(self, job: JobModel, queue: Queue) -> None:
             self._model.heartbeat(pipeline, self.job_monitoring_interval + 60)
             ttl = self.get_heartbeat_ttl(job)
 
-            queue.active_job_registry.add(pipeline, self.name, current_timestamp() + ttl, update_existing_only=False)
+            queue.active_job_registry.add(pipeline, job.name, current_timestamp() + ttl, update_existing_only=False)
             results = pipeline.execute()
             if results[2] == 1:
-                job.delete(self.connection)
+                job.delete(queue.connection)
 
     def execute_in_separate_process(self, job: JobModel, queue: Queue) -> None:
         """This is the entry point of the newly spawned job execution process.