Fixed: Queued jobs are not considered in deferring logic

.vscode/settings.json

@@ -40,9 +40,4 @@
     "cSpell.words": [
         "crowdsourcing"
     ],
-    "isort.args":["--profile", "black"],
-    "[python]": {
-        "editor.defaultFormatter": "ms-python.black-formatter",
-        "editor.formatOnSave": true
-    }
 }

cvat/apps/engine/backup.py

@@ -1,5 +1,5 @@
 # Copyright (C) 2021-2022 Intel Corporation
-# Copyright (C) 2022-2023 CVAT.ai Corporation
+# Copyright (C) 2022-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -15,8 +15,11 @@
 from zipfile import ZipFile
 from datetime import datetime
 from tempfile import NamedTemporaryFile
+from contextlib import suppress
 
 import django_rq
+from rq.command import send_stop_job_command
+from rq.exceptions import InvalidJobOperation
 from attr.converters import to_bool
 from django.conf import settings
 from django.db import transaction
@@ -967,6 +970,8 @@ def export(db_instance, request, queue_name):
             # in case the server is configured with ONE_RUNNING_JOB_IN_QUEUE_PER_USER
             # we have to enqueue dependent jobs after canceling one
             rq_job.cancel(enqueue_dependents=settings.ONE_RUNNING_JOB_IN_QUEUE_PER_USER)
+            with suppress(InvalidJobOperation):
+                send_stop_job_command(rq_job.connection, rq_job.id)
             rq_job.delete()
         else:
             if rq_job.is_finished:

cvat/apps/engine/utils.py

@@ -165,25 +165,18 @@ def define_dependent_job(
     if not should_be_dependent:
         return None
 
-    started_user_jobs = [
-        job
-        for job in queue.job_class.fetch_many(
-            queue.started_job_registry.get_job_ids(), queue.connection
-        )
-        if job and job.meta.get("user", {}).get("id") == user_id
-    ]
-    deferred_user_jobs = [
-        job
-        for job in queue.job_class.fetch_many(
-            queue.deferred_job_registry.get_job_ids(), queue.connection
-        )
-        # Since there is no cleanup implementation in DeferredJobRegistry,
-        # this registry can contain "outdated" jobs that weren't deleted from it
-        # but were added to another registry. Probably such situations can occur
-        # if there are active or deferred jobs when restarting the worker container.
-        if job and job.meta.get("user", {}).get("id") == user_id and job.is_deferred
-    ]
-    all_user_jobs = started_user_jobs + deferred_user_jobs
+    queues = [queue.deferred_job_registry, queue, queue.started_job_registry]
+    # Since there is no cleanup implementation in DeferredJobRegistry,
+    # this registry can contain "outdated" jobs that weren't deleted from it
+    # but were added to another registry. Probably such situations can occur
+    # if there are active or deferred jobs when restarting the worker container.
+    filters = [lambda job: job.is_deferred, lambda _: True, lambda _: True]
+    all_user_jobs = []
+    for q, f in zip(queues, filters):
+        job_ids = q.get_job_ids()
+        jobs = q.job_class.fetch_many(job_ids, q.connection)
+        jobs = filter(lambda job: job.meta.get("user", {}).get("id") == user_id and f(job), jobs)
+        all_user_jobs.extend(jobs)
 
     # prevent possible cyclic dependencies
     if rq_id:

cvat/apps/engine/views.py

@@ -1,5 +1,5 @@
 # Copyright (C) 2018-2022 Intel Corporation
-# Copyright (C) 2022-2023 CVAT.ai Corporation
+# Copyright (C) 2022-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
@@ -14,8 +14,11 @@
 from datetime import datetime
 from tempfile import NamedTemporaryFile
 from textwrap import dedent
+from contextlib import suppress
 
 import django_rq
+from rq.command import send_stop_job_command
+from rq.exceptions import InvalidJobOperation
 from attr.converters import to_bool
 from django.conf import settings
 from django.contrib.auth.models import User
@@ -2965,6 +2968,7 @@ def _export_annotations(
         tasks_update = list(map(lambda db_task: timezone.localtime(db_task.updated_date), db_instance.tasks.all()))
         last_instance_update_time = max(tasks_update + [last_instance_update_time])
 
+
     timestamp = datetime.strftime(last_instance_update_time, "%Y_%m_%d_%H_%M_%S")
     is_annotation_file = rq_id.startswith('export:annotations')
 
@@ -2975,6 +2979,8 @@ def _export_annotations(
             # in case the server is configured with ONE_RUNNING_JOB_IN_QUEUE_PER_USER
             # we have to enqueue dependent jobs after canceling one
             rq_job.cancel(enqueue_dependents=settings.ONE_RUNNING_JOB_IN_QUEUE_PER_USER)
+            with suppress(InvalidJobOperation):
+                send_stop_job_command(rq_job.connection, rq_job.id)
             rq_job.delete()
         else:
             if rq_job.is_finished:

cvat/rq_patching.py

@@ -1,16 +1,18 @@
-# Copyright (C) 2023 CVAT.ai Corporation
+# Copyright (C) 2023-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
 import traceback
+import sys
 from datetime import datetime
 from typing import Optional
 
 import rq.registry
 from rq.exceptions import AbandonedJobError, NoSuchJobError
 from rq.job import JobStatus
-from rq.utils import current_timestamp
+from rq.utils import current_timestamp, utcformat
 from rq.version import VERSION
+from rq.worker import StopRequested
 
 
 # NOTE: we should patch implementation of original method because
@@ -69,7 +71,22 @@ def custom_started_job_registry_cleanup(self, timestamp: Optional[float] = None)
 
     return job_ids
 
+def custom_hearbeat(self, timestamp: datetime, ttl: int, pipeline: Optional['Pipeline'] = None, xx: bool = False):
+    self.last_heartbeat = timestamp
+    connection = pipeline if pipeline is not None else self.connection
+    connection.hset(self.key, 'last_heartbeat', utcformat(self.last_heartbeat))
+
+    # the only difference with the default implementation
+    # is to avoid adding job to started job registry
+    # that is not clear why they adds a job to started_job_registry when the job has failed
+    # also, as far as I see they also commented this line in unrealeased (2.0) version
+    exc_info = sys.exc_info()
+    is_stopped_export_job = isinstance(exc_info[1], (StopRequested, SystemExit))
+    if not is_stopped_export_job:
+        self.started_job_registry.add(self, ttl, pipeline=pipeline, xx=xx)
+
 def update_started_job_registry_cleanup() -> None:
     # don't forget to check if the issue https://github.com/rq/rq/issues/2006 has been resolved in upstream
     assert VERSION == '1.16.0'
     rq.registry.StartedJobRegistry.cleanup = custom_started_job_registry_cleanup
+    rq.job.Job.heartbeat = custom_hearbeat

cvat/rqworker.py

@@ -1,16 +1,34 @@
 # Copyright (C) 2018-2022 Intel Corporation
-# Copyright (C) 2022-2023 CVAT.ai Corporation
+# Copyright (C) 2022-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
 import os
-
+import signal
 from rq import Worker
+from rq.worker import StopRequested
 
 import cvat.utils.remote_debugger as debug
 
+class CVATWorker(Worker):
+    # may be called from work horse's perform_job::except block
+    # or from parent's Worker::monitor_work_horse_process
+    # if parent process sees that work-horse is dead
+
+    # This modification ensures that jobs stopped intentionally
+    # do not get their status updated or placed in the failed registry
+    # as the main server code must delete them at all
+    def handle_job_failure(self, job, queue, **kwargs):
+        # pylint: disable=access-member-before-definition
+        if self._stopped_job_id == job.id:
+            self._stopped_job_id = None
+            self.set_current_job_id(None)
+        else:
+            super().handle_job_failure(job, queue, **kwargs)
+
 
-DefaultWorker = Worker
+
+DefaultWorker = CVATWorker
 
 
 class BaseDeathPenalty:
@@ -24,7 +42,7 @@ def __exit__(self, exc_type, exc_value, traceback):
         pass
 
 
-class SimpleWorker(Worker):
+class SimpleWorker(CVATWorker):
     """
     Allows to work with at most 1 worker thread. Useful for debugging.
     """
@@ -46,6 +64,27 @@ def execute_job(self, *args, **kwargs):
 
         return self.perform_job(*args, **kwargs)
 
+    def kill_horse(self, sig: signal.Signals = signal.SIGTERM):
+        # In debug mode we send SIGTERM instead of default SIGKILL
+        # Because SIGKILL is not handled (and can't be handled) by RQ code and it kills debug process from VSCode
+        # All three signals (SIGKILL, SIGTERM, SIGINT) are regularly used at RQ code
+        super().kill_horse(sig)
+
+    def handle_exception(self, *args, **kwargs):
+        # In production environment it sends SIGKILL to work horse process and this method never called
+        # But for development we overrided the signal and it sends SIGTERM to the process
+        # This modification ensures that exceptions are handled differently
+        # when the job is stopped intentionally, avoiding incorrect exception handling.
+
+        # PROBLEM: default "handle_exception" code saves meta with exception
+        # It leads to bugs:
+        # - we create another job with the same ID in the server process
+        # - when exception message is saved in worker code, it also saves outdated datetime value as part of meta information
+        # - this outdated value then used in server code
+        is_stopped_export_job = isinstance(args[2], (StopRequested, SystemExit))
+        if not is_stopped_export_job:
+            super().handle_exception(*args, **kwargs)
+
 
 if debug.is_debugging_enabled():
     class RemoteDebugWorker(SimpleWorker):