ITEP-36123: Server returns 500 error when invoking the same job twice (#425)

Author: warrkan

interactive_ai/services/director/app/communication/controllers/training_controller.py

@@ -8,6 +8,7 @@
 
 from communication.data_validator import TrainingRestValidator
 from communication.exceptions import (
+    JobDuplicateFoundException,
     JobInsufficientBalanceException,
     NotEnoughDatasetItemsException,
     NotEnoughSpaceHTTPException,
@@ -26,7 +27,7 @@
 from geti_fastapi_tools.exceptions import BadRequestException
 from geti_telemetry_tools import unified_tracing
 from geti_types import ID, DatasetStorageIdentifier
-from grpc_interfaces.job_submission.client import InsufficientBalanceException
+from grpc_interfaces.job_submission.client import DuplicateFoundException, InsufficientBalanceException
 from iai_core.algorithms import ModelTemplateList
 from iai_core.entities.annotation_scene_state import AnnotationState
 from iai_core.entities.project import Project
@@ -201,6 +202,9 @@ def _submit_train_job(project: Project, task_training_config: TrainingConfig, au
         except InsufficientBalanceException:
             logger.error("Insufficient balance for job execution")
             raise JobInsufficientBalanceException("Insufficient balance for job execution")
+        except DuplicateFoundException:
+            logger.error("Duplicate running job has been found")
+            raise JobDuplicateFoundException("Duplicate running job has been found")
 
     @staticmethod
     def _get_task_readiness_status_by_project(

interactive_ai/services/director/app/communication/exceptions.py

@@ -172,6 +172,22 @@ def __init__(self, message: str) -> None:
         )
 
 
+class JobDuplicateFoundException(GetiBaseException):
+    """
+    This error can be raised if there is a running duplicate job is found during submission and job policy is set
+    to REJECT
+
+    :param message: str containing a custom error message
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(
+            message=message,
+            error_code="job_duplicate_found",
+            http_status=http.HTTPStatus.PRECONDITION_FAILED,
+        )
+
+
 class NotReadyForTrainingException(GetiBaseException):
     """
     Exception raised when a manual training trigger is received for a task that is not

libs/grpc_interfaces/src/grpc_interfaces/job_submission/client.py

@@ -44,6 +44,10 @@ class InsufficientBalanceException(CommunicationError):
     """Exception for when organization doesn't have enough credits for job submission."""
 
 
+class DuplicateFoundException(CommunicationError):
+    """Exception for when duplicate policy is set to REJECT and duplicate job is found during submission."""
+
+
 def grpc_requester(acquire_lock: bool = True):  # noqa: ANN201
     """
     Decorator for methods of GRPCJobsClient which handles GRPC requests.
@@ -357,6 +361,8 @@ def handle_grpc_error(self, error: RpcError) -> None:
 
         if "Insufficient balance" in details:
             raise InsufficientBalanceException(details) from error
+        if "An identical job was found in the job queue" in details:
+            raise DuplicateFoundException(details) from error
 
         if not details:
             details = "Request failed due to unknown issues. Please try again later."