Merge pull request #1640 from roboflow/feature/custom-error-handler-in-workflows

Add custom error handlers in Workflows

Author: PawelPeczek-Roboflow

docs/workflows/execution_engine_changelog.md

@@ -2,6 +2,36 @@
 
 Below you can find the changelog for Execution Engine.
 
+## Execution Engine `v1.7.0` | inference `v0.59.0`
+
+!!! warning "Breaking change regarding step errors in workflows"
+  
+    To fix a bug related to invalid HTTP responses codes in `inference-server` handling Workflows execution requests 
+    we needed to alter the default mechanism responsible for handling errors in Execution Engine. As a result of change,
+    effective immediately on Roboflow Hosted Platform and in `inference>=0.59.0`, Workflow blocks interacting with 
+    Roboflow platform which fails due to client misconfiguration (invalid Roboflow API key, invalid model ID, etc.) 
+    instead of raising `StepExecutionError` (and HTTP 500 response from the server) will raise 
+    `ClientCausedStepExecutionError` (and relevant HTTP response codes, such as 400, 401, 403, 404).
+
+List of scenarios affected with the change:
+
+* Block using Roboflow model defines invalid model ID - now will raise `ClientCausedStepExecutionError` with status code 400
+
+* Block using Roboflow model defines invalid API key - now will raise `ClientCausedStepExecutionError` with status code 401
+
+* Block using Roboflow model defines invalid API key or missing valid key with scpe to access resource - now will raise 
+`ClientCausedStepExecutionError` with status code 403
+
+* Block using Roboflow model defines model which does not exist - now will raise 
+`ClientCausedStepExecutionError` with status code 404
+
+
+!!! Note "Bringing back `legacy` error handling"
+    
+    It is possible to bring back the legacy behaviour of error handler if needed, which may be halpful in transition 
+    period - all it takes is setting environmental variable `DEFAULT_WORKFLOWS_STEP_ERROR_HANDLER=legacy`.
+
+
 ## Execution Engine `v1.6.0` | inference `v0.53.0`
 
 !!! Note "Change may require attention"
@@ -552,5 +582,3 @@ include a new `video_metadata` property. This property can be optionally set in
 a default value with reasonable defaults will be used. To simplify metadata manipulation within blocks, we have 
 introduced two new class methods: `WorkflowImageData.copy_and_replace(...)` and `WorkflowImageData.create_crop(...)`. 
 For more details, refer to the updated [`WoorkflowImageData` usage guide](/workflows/internal_data_types.md#workflowimagedata).
-
-

inference/core/exceptions.py

@@ -151,6 +151,10 @@ class RoboflowAPINotAuthorizedError(RoboflowAPIUnsuccessfulRequestError):
     pass
 
 
+class RoboflowAPIForbiddenError(RoboflowAPIUnsuccessfulRequestError):
+    pass
+
+
 class RoboflowAPINotNotFoundError(RoboflowAPIUnsuccessfulRequestError):
     pass
 

inference/core/interfaces/http/error_handlers.py

@@ -22,6 +22,7 @@
     PostProcessingError,
     PreProcessingError,
     RoboflowAPIConnectionError,
+    RoboflowAPIForbiddenError,
     RoboflowAPINotAuthorizedError,
     RoboflowAPINotNotFoundError,
     RoboflowAPITimeoutError,
@@ -45,6 +46,7 @@
     OperationTypeNotRecognisedError,
 )
 from inference.core.workflows.errors import (
+    ClientCausedStepExecutionError,
     DynamicBlockError,
     ExecutionGraphStructureError,
     InvalidReferenceTargetError,
@@ -187,6 +189,16 @@ def wrapped_route(*args, **kwargs):
                     "to learn how to retrieve one."
                 },
             )
+        except RoboflowAPIForbiddenError as error:
+            logger.exception("%s: %s", type(error).__name__, error)
+            resp = JSONResponse(
+                status_code=403,
+                content={
+                    "message": "Unauthorized access to roboflow API - check API key and make sure the key is valid and "
+                    "have required scopes. Visit https://docs.roboflow.com/api-reference/authentication#retrieve-an-api-key "
+                    "to learn how to retrieve one."
+                },
+            )
         except RoboflowAPINotNotFoundError as error:
             logger.exception("%s: %s", type(error).__name__, error)
             resp = JSONResponse(
@@ -284,6 +296,24 @@ def wrapped_route(*args, **kwargs):
                     "message": "Timeout when attempting to connect to Roboflow API."
                 },
             )
+        except ClientCausedStepExecutionError as error:
+            logger.exception("%s: %s", type(error).__name__, error)
+            content = WorkflowErrorResponse(
+                message=str(error.public_message),
+                error_type=error.__class__.__name__,
+                context=str(error.context),
+                inner_error_type=str(error.inner_error_type),
+                inner_error_message=str(error.inner_error),
+                blocks_errors=[
+                    WorkflowBlockError(
+                        block_id=error.block_id,
+                    ),
+                ],
+            )
+            resp = JSONResponse(
+                status_code=error.status_code,
+                content=content.model_dump(),
+            )
         except StepExecutionError as error:
             logger.exception("%s: %s", type(error).__name__, error)
             content = WorkflowErrorResponse(
@@ -462,6 +492,16 @@ async def wrapped_route(*args, **kwargs):
                     "to learn how to retrieve one."
                 },
             )
+        except RoboflowAPIForbiddenError as error:
+            logger.exception("%s: %s", type(error).__name__, error)
+            resp = JSONResponse(
+                status_code=403,
+                content={
+                    "message": "Unauthorized access to roboflow API - check API key and make sure the key is valid and "
+                    "have required scopes. Visit https://docs.roboflow.com/api-reference/authentication#retrieve-an-api-key "
+                    "to learn how to retrieve one."
+                },
+            )
         except RoboflowAPINotNotFoundError as error:
             logger.exception("%s: %s", type(error).__name__, error)
             resp = JSONResponse(
@@ -559,6 +599,24 @@ async def wrapped_route(*args, **kwargs):
                     "message": "Timeout when attempting to connect to Roboflow API."
                 },
             )
+        except ClientCausedStepExecutionError as error:
+            logger.exception("%s: %s", type(error).__name__, error)
+            content = WorkflowErrorResponse(
+                message=str(error.public_message),
+                error_type=error.__class__.__name__,
+                context=str(error.context),
+                inner_error_type=str(error.inner_error_type),
+                inner_error_message=str(error.inner_error),
+                blocks_errors=[
+                    WorkflowBlockError(
+                        block_id=error.block_id,
+                    ),
+                ],
+            )
+            resp = JSONResponse(
+                status_code=error.status_code,
+                content=content.model_dump(),
+            )
         except StepExecutionError as error:
             logger.exception("%s: %s", type(error).__name__, error)
             content = WorkflowErrorResponse(

inference/core/roboflow_api.py

@@ -53,6 +53,7 @@
     MissingDefaultModelError,
     RetryRequestError,
     RoboflowAPIConnectionError,
+    RoboflowAPIForbiddenError,
     RoboflowAPIIAlreadyAnnotatedError,
     RoboflowAPIIAnnotationRejectionError,
     RoboflowAPIImageUploadRejectionError,
@@ -97,6 +98,12 @@ def raise_from_lambda(
         "Unauthorized access to roboflow API - check API key. Visit "
         "https://docs.roboflow.com/api-reference/authentication#retrieve-an-api-key to learn how to retrieve one.",
     ),
+    403: lambda e: raise_from_lambda(
+        e,
+        RoboflowAPIForbiddenError,
+        "Unauthorized access to roboflow API - check API key regarding correctness and required scopes. Visit "
+        "https://docs.roboflow.com/api-reference/authentication#retrieve-an-api-key to learn how to retrieve one.",
+    ),
     404: lambda e: raise_from_lambda(
         e, RoboflowAPINotNotFoundError, NOT_FOUND_ERROR_MESSAGE
     ),

inference/core/version.py

@@ -1,4 +1,4 @@
-__version__ = "0.58.3"
+__version__ = "0.59.0"
 
 
 if __name__ == "__main__":

inference/core/workflows/core_steps/sinks/onvif_movement/v1.py

@@ -17,13 +17,6 @@
 from inference.core import logger
 from inference.core.utils.function import experimental
 from inference.core.workflows.core_steps.common.entities import StepExecutionMode
-from inference.core.workflows.core_steps.common.query_language.entities.operations import (
-    AllOperationsType,
-)
-from inference.core.workflows.core_steps.common.query_language.operations.core import (
-    build_operations_chain,
-)
-from inference.core.workflows.errors import StepExecutionError, WorkflowError
 from inference.core.workflows.execution_engine.entities.base import OutputDefinition
 from inference.core.workflows.execution_engine.entities.types import (
     BOOLEAN_KIND,
@@ -403,9 +396,7 @@ async def configure_async(self):
         Doesn't currently run in init since it needs to be awaited
         """
         if not self.camera:
-            raise StepExecutionError(
-                f"Tried to configure camera, but camera was not created"
-            )
+            raise ValueError(f"Tried to configure camera, but camera was not created")
         if self._has_config_error:
             return
 

inference/core/workflows/errors.py

@@ -173,6 +173,22 @@ def __init__(
         self.block_type = block_type
 
 
+class ClientCausedStepExecutionError(WorkflowExecutionEngineError):
+    def __init__(
+        self,
+        block_id: str,
+        status_code: int,
+        public_message: str,
+        context: str,
+        inner_error: Optional[Exception] = None,
+    ):
+        super().__init__(
+            public_message=public_message, context=context, inner_error=inner_error
+        )
+        self.block_id = block_id
+        self.status_code = status_code
+
+
 class ExecutionEngineRuntimeError(WorkflowExecutionEngineError):
     pass
 

inference/core/workflows/execution_engine/core.py

@@ -1,5 +1,5 @@
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Dict, List, Optional, Type
+from typing import Any, Callable, Dict, List, Optional, Type, Union
 
 from packaging.specifiers import SpecifierSet
 from packaging.version import Version
@@ -14,6 +14,7 @@
 )
 from inference.core.workflows.execution_engine.profiling.core import WorkflowsProfiler
 from inference.core.workflows.execution_engine.v1.core import (
+    DEFAULT_WORKFLOWS_STEP_ERROR_HANDLER,
     EXECUTION_ENGINE_V1_VERSION,
     ExecutionEngineV1,
 )
@@ -39,6 +40,9 @@ def init(
         workflow_id: Optional[str] = None,
         profiler: Optional[WorkflowsProfiler] = None,
         executor: Optional[ThreadPoolExecutor] = None,
+        step_error_handler: Optional[
+            Union[str, Callable[[str, Exception], None]]
+        ] = DEFAULT_WORKFLOWS_STEP_ERROR_HANDLER,
     ) -> "ExecutionEngine":
         requested_engine_version = retrieve_requested_execution_engine_version(
             workflow_definition=workflow_definition,
@@ -54,6 +58,7 @@ def init(
             workflow_id=workflow_id,
             profiler=profiler,
             executor=executor,
+            step_error_handler=step_error_handler,
         )
         return cls(engine=engine)
 

inference/core/workflows/execution_engine/entities/engine.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Union
 
 from inference.core.workflows.execution_engine.profiling.core import WorkflowsProfiler
 
@@ -18,6 +18,9 @@ def init(
         workflow_id: Optional[str] = None,
         profiler: Optional[WorkflowsProfiler] = None,
         executor: Optional[ThreadPoolExecutor] = None,
+        step_error_handler: Optional[
+            Union[str, Callable[[str, Exception], None]]
+        ] = None,
     ) -> "BaseExecutionEngine":
         pass
 

inference/core/workflows/execution_engine/v1/core.py

@@ -1,9 +1,11 @@
+import os
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Union
 
 from packaging.version import Version
 
 from inference.core.logger import logger
+from inference.core.workflows.errors import WorkflowEnvironmentConfigurationError
 from inference.core.workflows.execution_engine.entities.engine import (
     BaseExecutionEngine,
 )
@@ -22,8 +24,21 @@
 from inference.core.workflows.execution_engine.v1.executor.runtime_input_validator import (
     validate_runtime_input,
 )
+from inference.core.workflows.execution_engine.v1.step_error_handlers import (
+    extended_roboflow_errors_handler,
+    legacy_step_error_handler,
+)
+
+EXECUTION_ENGINE_V1_VERSION = Version("1.7.0")
 
-EXECUTION_ENGINE_V1_VERSION = Version("1.6.0")
+DEFAULT_WORKFLOWS_STEP_ERROR_HANDLER = os.getenv(
+    "DEFAULT_WORKFLOWS_STEP_ERROR_HANDLER", "extended_roboflow_errors"
+)
+
+REGISTERED_STEP_ERROR_HANDLERS = {
+    "legacy": legacy_step_error_handler,
+    "extended_roboflow_errors": extended_roboflow_errors_handler,
+}
 
 
 class ExecutionEngineV1(BaseExecutionEngine):
@@ -38,10 +53,21 @@ def init(
         workflow_id: Optional[str] = None,
         profiler: Optional[WorkflowsProfiler] = None,
         executor: Optional[ThreadPoolExecutor] = None,
+        step_error_handler: Optional[
+            Union[str, Callable[[str, Exception], None]]
+        ] = DEFAULT_WORKFLOWS_STEP_ERROR_HANDLER,
     ) -> "ExecutionEngineV1":
         if init_parameters is None:
             init_parameters = {}
-
+        if isinstance(step_error_handler, str):
+            if step_error_handler not in REGISTERED_STEP_ERROR_HANDLERS:
+                raise WorkflowEnvironmentConfigurationError(
+                    public_message=f"Execution engine was initialised with step_error_handler='{step_error_handler}' "
+                    f"which is not registered. Supported values: "
+                    f"{list(REGISTERED_STEP_ERROR_HANDLERS.keys())}",
+                    context="workflow_compilation | engine_initialisation",
+                )
+            step_error_handler = REGISTERED_STEP_ERROR_HANDLERS[step_error_handler]
         init_parameters["dynamic_workflows_blocks.api_key"] = init_parameters.get(
             "dynamic_workflows_blocks.api_key",
             init_parameters.get("workflows_core.api_key"),
@@ -63,6 +89,7 @@ def init(
             workflow_id=workflow_id,
             internal_id=workflow_definition.get("id"),
             executor=executor,
+            step_error_handler=step_error_handler,
         )
 
     def __init__(
@@ -74,6 +101,7 @@ def __init__(
         workflow_id: Optional[str] = None,
         internal_id: Optional[str] = None,
         executor: Optional[ThreadPoolExecutor] = None,
+        step_error_handler: Optional[Callable[[str, Exception], None]] = None,
     ):
         self._compiled_workflow = compiled_workflow
         self._max_concurrent_steps = max_concurrent_steps
@@ -82,6 +110,7 @@ def __init__(
         self._profiler = profiler
         self._internal_id = internal_id
         self._executor = executor
+        self._step_error_handler = step_error_handler
 
     def run(
         self,
@@ -121,6 +150,7 @@ def run(
             serialize_results=serialize_results,
             profiler=self._profiler,
             executor=self._executor,
+            step_error_handler=self._step_error_handler,
         )
         self._profiler.end_workflow_run()
         return result