Merge branch 'master' into save_safety

Author: sliverc

.github/workflows/test-workflow.yml

@@ -0,0 +1,46 @@
+name: Test workflow
+
+on:
+  workflow_dispatch:
+    inputs:
+      environment:
+        description: 'Environment to run the workflow in'
+        required: true
+        type: environment
+      reason:
+        description: 'Reason for running the workflow'
+        required: true
+        type: string
+      environment-optional:
+        description: 'Environment to run the workflow in'
+        required: false
+        type: environment
+      choices-input-optional:
+        type: choice
+        required: false
+        options:
+          - option 1
+          - option 2
+        description: "Choice input"
+      number-input-optional:
+        type: number
+        required: false
+        description: "Number input"
+
+jobs:
+  ruff:
+    runs-on: ubuntu-latest
+    name: "ruff on code"
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@v5
+        with:
+          persist-credentials: false
+      - name: print all inputs
+        run: |
+          echo "environment: ${{ github.event.inputs.environment }}"
+          echo "reason: ${{ github.event.inputs.reason }}"
+          echo "environment-optional: ${{ github.event.inputs.environment-optional }}"
+          echo "choices-input-optional: ${{ github.event.inputs.choices-input-optional }}"
+          echo "number-input-optional: ${{ github.event.inputs.number-input-optional }}"

docs/changelog.md

@@ -1,14 +1,12 @@
 # Changelog
 
-## v4.0.8 🌈
+## v4.0.7 🌈
 
 ### 🐛 Bug Fixes
 
 - Fix: Scheduled jobs fail to execute: {'scheduled_time': ['Scheduled time must be in the future']} #297
 - Fix Error with deserialize JobModel due to multi-processing #291
 
-## v4.0.7 🌈
-
 ### 🧰 Maintenance
 
 - Improve GitHub workflow performance using ruff action @DhavalGojiya #294

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "django-tasks-scheduler"
-version = "4.0.8"
+version = "4.0.7"
 description = "An async job scheduler for django using redis/valkey brokers"
 authors = [{ name = "Daniel Moran", email = "[email protected]" }]
 requires-python = ">=3.10"

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/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/job.py

@@ -128,7 +128,7 @@ 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.timeout)
         self.save(connection=connection)
 
     def after_execution(

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/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/tests/test_worker/test_worker_commands.py

@@ -1,16 +1,26 @@
 import json
 from threading import Thread
 from time import sleep
+from unittest import mock
 
 from scheduler.helpers.queues import get_queue
 from scheduler.tests.jobs import test_job, two_seconds_job
 from ..test_views.base import BaseTestCase
+from ...helpers.callback import Callback
 from ...redis_models import JobModel, JobStatus, WorkerModel
 from ...worker import create_worker
 from ...worker.commands import send_command, StopJobCommand
 from ...worker.commands.suspend_worker import SuspendWorkCommand
 
 
+def _callback_func():
+    pass
+
+
+def callback_func():
+    pass
+
+
 class WorkerCommandsTest(BaseTestCase):
     def test_stop_worker_command__green(self):
         # Arrange
@@ -45,11 +55,12 @@ def test_stop_worker_command__bad_worker_name(self):
         job = JobModel.get(job.name, connection=queue.connection)
         self.assertFalse(job.is_queued)
 
-    def test_stop_job_command__success(self):
+    @mock.patch("scheduler.redis_models.job.JobModel.call_stopped_callback")
+    def test_stop_job_command__success(self, mock_stopped_callback):
         # Arrange
         worker_name = "test"
         queue = get_queue("default")
-        job = queue.create_and_enqueue_job(two_seconds_job)
+        job = queue.create_and_enqueue_job(two_seconds_job, on_stopped=Callback(callback_func))
         self.assertTrue(job.is_queued)
         worker = create_worker("default", name=worker_name, burst=True, with_scheduler=False)
         worker.bootstrap()
@@ -70,3 +81,4 @@ def test_stop_job_command__success(self):
         self.assertIsNone(worker.current_job_name)
         self.assertEqual(job.status, JobStatus.STOPPED)
         t.join()
+        mock_stopped_callback.assert_called()

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
@@ -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,17 +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."""
-        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, self.name, current_timestamp() + ttl, update_existing_only=False)
-            results = pipeline.execute()
-            if results[2] == 1:
-                job.delete(self.connection)
-
     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.
@@ -785,10 +774,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}`...")