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

@@ -16,6 +16,7 @@
 from textwrap import dedent
 
 import django_rq
+from rq.command import send_stop_job_command
 from attr.converters import to_bool
 from django.conf import settings
 from django.contrib.auth.models import User
@@ -2965,16 +2966,22 @@ 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')
 
     if rq_job:
         rq_request = rq_job.meta.get('request', None)
         request_time = rq_request.get('timestamp', None) if rq_request else None
         if request_time is None or request_time < last_instance_update_time:
-            # 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)
+            if rq_job.get_status() == "started":
+                send_stop_job_command(rq_job.connection, rq_job.id)
+            else:
+                # 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
+                )
             rq_job.delete()
         else:
             if rq_job.is_finished:

cvat/rqworker.py

@@ -1,16 +1,29 @@
 # 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):
+    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:
+            # the job was stopped intentionally, we do not need update its status or put it into failed registry
+            # in our case the job is immediately removed after stop request
+            super().handle_job_failure(job, queue, **kwargs)
+
 
-DefaultWorker = Worker
+
+DefaultWorker = CVATWorker
 
 
 class BaseDeathPenalty:
@@ -24,7 +37,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 +59,22 @@ def execute_job(self, *args, **kwargs):
 
         return self.perform_job(*args, **kwargs)
 
+    def kill_horse(self, sig: signal.Signals = signal.SIGTERM):
+        # Send SIGTERM instead of default SIGKILL in debug mode as SIGKILL can't be handled
+        # to prevent killing debug process (rq code handles SIGTERM properly)
+        super().kill_horse(sig)
+
+    def handle_exception(self, *args, **kwargs):
+        # on production it sends SIGKILL to work horse process
+        # but for development we overrided it and it sends SIGTERM to the process
+        # we need to prevent exception handling as the process killed intentilnally
+
+        # moreover default code saves meta with exception
+        # and rewrites request datetime in meta with old value, as new job with the same ID may aldeady been created in a new process
+        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):