Repository: AgnostiqHQ/covalent (language: python)

Issue:
Errors when instantiating executors can cause workflows to hang
When `_run_task` encounters an error when instantiating an executor, the workflow can hang. This occurs for the following reason.

There are [two sources of failure](https://github.com/AgnostiqHQ/covalent/blob/b45fef20b109607173752f447b8e7ed761620338/covalent_dispatcher/_core/execution.py#L246) in `_run_task`: `Executor.execute()`, and instantiating executors. Only the former is handled correctly.

When electrons raise exceptions in `_run_task()`, the exception is caught by `_run_task` and returned in the node result. The failed node result triggers a [handler](https://github.com/AgnostiqHQ/covalent/blob/b45fef20b109607173752f447b8e7ed761620338/covalent_dispatcher/_core/execution.py#L421) that "poisons" the task queue to [signal the dispatcher](https://github.com/AgnostiqHQ/covalent/blob/develop/covalent_dispatcher/_core/execution.py#L626) to stop submitting tasks and clean up.

Unfortunately, exceptions from instantiating executors are simply [re-raised](https://github.com/AgnostiqHQ/covalent/blob/b45fef20b109607173752f447b8e7ed761620338/covalent_dispatcher/_core/execution.py#L288). This behavior is incorrect because it bypasses [all node handlers](https://github.com/AgnostiqHQ/covalent/blob/b45fef20b109607173752f447b8e7ed761620338/covalent_dispatcher/_core/execution.py#L463) and will cause the dispatcher to listen on the `tasks_queue` indefinitely if additional tasks remain to be submitted.

The solution is to always return a `node_result` from `_run_task`; all exceptions must be caught and handled.





Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2c806165b..ac90a4bc8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Ability to use terminal on the GUI.
 
+### Fixed
+
+- Exceptions when instantiating executors are handled
+
 ## [0.202.0] - 2022-10-11
 
 ### Authors
diff --git a/covalent_dispatcher/_core/execution.py b/covalent_dispatcher/_core/execution.py
index f0bf13942..3b327904c 100644
--- a/covalent_dispatcher/_core/execution.py
+++ b/covalent_dispatcher/_core/execution.py
@@ -205,11 +205,12 @@ async def _dispatch_sync_sublattice(
         short_name, object_dict = workflow_executor
 
         if short_name == "client":
+            app_log.error("No executor selected for dispatching sublattices")
             raise RuntimeError("No executor selected for dispatching sublattices")
 
     except Exception as ex:
         app_log.debug(f"Exception when trying to determine sublattice executor: {ex}")
-        raise ex
+        return None
 
     sub_dispatch_inputs = {"args": [serialized_callable], "kwargs": inputs["kwargs"]}
     for arg in inputs["args"]:
@@ -233,14 +234,17 @@ async def _dispatch_sync_sublattice(
     )
 
     res = await fut
-    json_sublattice = json.loads(res["output"].json)
+    if res["status"] == Result.COMPLETED:
+        json_sublattice = json.loads(res["output"].json)
 
-    sub_result_object = initialize_result_object(
-        json_sublattice, parent_result_object, parent_electron_id
-    )
-    app_log.debug(f"Sublattice dispatch id: {sub_result_object.dispatch_id}")
+        sub_result_object = initialize_result_object(
+            json_sublattice, parent_result_object, parent_electron_id
+        )
+        app_log.debug(f"Sublattice dispatch id: {sub_result_object.dispatch_id}")
 
-    return await run_workflow(sub_result_object)
+        return await run_workflow(sub_result_object)
+    else:
+        return None
 
 
 async def _run_task(
@@ -284,8 +288,14 @@ async def _run_task(
         executor = _executor_manager.get_executor(short_name)
         executor.from_dict(object_dict)
     except Exception as ex:
-        app_log.debug(f"Exception when trying to determine executor: {ex}")
-        raise ex
+        app_log.debug(f"Exception when trying to instantiate executor: {ex}")
+        node_result = generate_node_result(
+            node_id=node_id,
+            end_time=datetime.now(timezone.utc),
+            status=Result.FAILED,
+            error="".join(traceback.TracebackException.from_exception(ex).format()),
+        )
+        return node_result
 
     # run the task on the executor and register any failures
     try:
@@ -786,7 +796,6 @@ async def run_workflow(result_object: Result) -> Result:
             result_object.dispatch_id,
             "".join(traceback.TracebackException.from_exception(ex).format()),
         )
-        raise
 
     return result_object
 


Test patch:
diff --git a/tests/covalent_dispatcher_tests/_core/execution_test.py b/tests/covalent_dispatcher_tests/_core/execution_test.py
index ec07f4232..d5a6b0514 100644
--- a/tests/covalent_dispatcher_tests/_core/execution_test.py
+++ b/tests/covalent_dispatcher_tests/_core/execution_test.py
@@ -715,31 +715,23 @@ def sub_workflow(x):
 
     # Check handling of invalid workflow executors
 
-    try:
-        sub_result = await _dispatch_sync_sublattice(
-            parent_result_object=result_object,
-            parent_electron_id=1,
-            inputs=inputs,
-            serialized_callable=serialized_callable,
-            workflow_executor=["client", {}],
-        )
-        assert False
-
-    except RuntimeError:
-        pass
-
-    try:
-        sub_result = await _dispatch_sync_sublattice(
-            parent_result_object=result_object,
-            parent_electron_id=1,
-            inputs=inputs,
-            serialized_callable=serialized_callable,
-            workflow_executor=["fake_executor", {}],
-        )
-        assert False
+    sub_result = await _dispatch_sync_sublattice(
+        parent_result_object=result_object,
+        parent_electron_id=1,
+        inputs=inputs,
+        serialized_callable=serialized_callable,
+        workflow_executor=["client", {}],
+    )
+    assert sub_result is None
 
-    except ValueError:
-        pass
+    sub_result = await _dispatch_sync_sublattice(
+        parent_result_object=result_object,
+        parent_electron_id=1,
+        inputs=inputs,
+        serialized_callable=serialized_callable,
+        workflow_executor=["fake_executor", {}],
+    )
+    assert sub_result is None
 
 
 @pytest.mark.asyncio
@@ -815,3 +807,29 @@ async def test_run_task_sublattice_handling(test_db, mocker):
     )
 
     assert node_result["status"] == Result.FAILED
+
+
+@pytest.mark.asyncio
+async def test_run_task_executor_exception_handling(mocker):
+    """Test that exceptions from initializing executors are caught"""
+
+    result_object = get_mock_result()
+    inputs = {"args": [], "kwargs": {}}
+    mock_get_executor = mocker.patch(
+        "covalent_dispatcher._core.execution._executor_manager.get_executor",
+        side_effect=Exception(),
+    )
+
+    node_result = await _run_task(
+        result_object=result_object,
+        node_id=1,
+        inputs=inputs,
+        serialized_callable=None,
+        selected_executor=["nonexistent", {}],
+        call_before=[],
+        call_after=[],
+        node_name=sublattice_prefix,
+        workflow_executor=["local", {}],
+    )
+
+    assert node_result["status"] == Result.FAILED