Repository: dbt-labs/dbt-core
Pull request: Handle exceptions during node execution more elegantly.
resolves #9583

### Problem

Because of a bug in Python's thread pool mechanism, an uncaught exception on dbt's task threads would cause it to hang.

### Solution

Eliminate a follow-up exception that caused a top-level exception to escape.

### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me  
- [x] I have run this code in development and it appears to resolve the stated issue  
- [x] This PR includes tests, or tests are not required/relevant for this PR
- [x] This PR has no interface changes (e.g. macros, cli, logs, json artifacts, config files, adapter interface, etc) or this PR has already received feedback and approval from Product or DX
- [x] This PR includes [type annotations](https://docs.python.org/3/library/typing.html) for new and modified functions


Problem statement:
Do Not Hang on Worker Thread Exceptions
### Is this a new bug in dbt-core?

- [X] I believe this is a new bug in dbt-core
- [X] I have searched the existing issues, and I could not find an existing issue for this bug

### Current Behavior

For a while now, when there was an exception during the execution of one of dbt's implementations of BaseRunner, dbt would hang rather than crash or report the exception message.

### Expected Behavior

When a node fails to execute with an exception, we should record the failure as we would any other node error, and proceed normally.

Hints from issue discussion:

Commit message: Add task documentation.

Patch to review:
diff --git a/.changes/unreleased/Fixes-20240216-145632.yaml b/.changes/unreleased/Fixes-20240216-145632.yaml
new file mode 100644
index 00000000000..a02027f66a5
--- /dev/null
+++ b/.changes/unreleased/Fixes-20240216-145632.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Tighten exception handling to avoid worker thread hangs.
+time: 2024-02-16T14:56:32.858967-05:00
+custom:
+  Author: peterallenwebb
+  Issue: "9583"
diff --git a/core/dbt/task/README.md b/core/dbt/task/README.md
index 9de939e4cc4..2b32f5dbfa8 100644
--- a/core/dbt/task/README.md
+++ b/core/dbt/task/README.md
@@ -1 +1,43 @@
 # Task README
+
+### Task Hierarchy
+```
+BaseTask
+ ┣ CleanTask
+ ┣ ConfiguredTask
+ ┃ ┣ GraphRunnableTask
+ ┃ ┃  ┣ CloneTask
+ ┃ ┃  ┣ CompileTask
+ ┃ ┃  ┃ ┣ GenerateTask
+ ┃ ┃  ┃ ┣ RunTask
+ ┃ ┃  ┃ ┃ ┣ BuildTask
+ ┃ ┃  ┃ ┃ ┣ FreshnessTask
+ ┃ ┃  ┃ ┃ ┣ SeedTask
+ ┃ ┃  ┃ ┃ ┣ SnapshotTask
+ ┃ ┃  ┃ ┃ ┗ TestTask 
+ ┃ ┃  ┃ ┗ ShowTask
+ ┃ ┃  ┗ ListTask
+ ┃ ┣ RetryTask 
+ ┃ ┣ RunOperationTask
+ ┃ ┗ ServeTask
+ ┣ DebugTask
+ ┣ DepsTask
+ ┗ InitTask
+```
+
+### Runner Hierarchy
+```
+BaseRunner
+ ┣ CloneRunner
+ ┣ CompileRunner
+ ┃ ┣ GenericSqlRunner
+ ┃ ┃  ┣ SqlCompileRunner
+ ┃ ┃  ┗ SqlExecuteRunner
+ ┃ ┣ ModelRunner
+ ┃ ┃ ┣ SeedRunner
+ ┃ ┃ ┗ SnapshotRunner
+ ┃ ┣ ShowRunner
+ ┃ ┗ TestRunner
+ ┣ FreshnessRunner
+ ┗ SavedQueryRunner
+```
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index d44f88cfc0b..80432050d71 100644
--- a/core/dbt/task/runnable.py
+++ b/core/dbt/task/runnable.py
@@ -22,7 +22,7 @@
 from dbt.contracts.state import PreviousState
 from dbt_common.events.contextvars import log_contextvars, task_contextvars
 from dbt_common.events.functions import fire_event, warn_or_error
-from dbt_common.events.types import Formatting
+from dbt_common.events.types import Formatting, Note
 from dbt.events.types import (
     LogCancelLine,
     DefaultSelector,
@@ -221,19 +221,36 @@ def call_runner(self, runner: BaseRunner) -> RunResult:
                     )
                 )
             status: Dict[str, str] = {}
+            result = None
+            thread_exception = None
             try:
                 result = runner.run_with_hooks(self.manifest)
-            except Exception as exc:
-                raise DbtInternalError(f"Unable to execute node: {exc}")
+            except Exception as e:
+                thread_exception = e
             finally:
                 finishctx = TimestampNamed("finished_at")
                 with finishctx, DbtModelState(status):
-                    fire_event(
-                        NodeFinished(
-                            node_info=runner.node.node_info,
-                            run_result=result.to_msg_dict(),
+                    if result is not None:
+                        fire_event(
+                            NodeFinished(
+                                node_info=runner.node.node_info,
+                                run_result=result.to_msg_dict(),
+                            )
                         )
-                    )
+                    else:
+                        msg = f"Exception on worker thread. {thread_exception}"
+                        fire_event(Note(msg=msg))
+                        result = RunResult(
+                            status=RunStatus.Error,  # type: ignore
+                            timing=[],
+                            thread_id="",
+                            execution_time=0.0,
+                            adapter_response={},
+                            message=msg,
+                            failures=None,
+                            node=runner.node,
+                        )
+
             # `_event_status` dict is only used for logging.  Make sure
             # it gets deleted when we're done with it, except for unit tests
             if not runner.node.resource_type == NodeType.Unit:


Code review comments:
diff_hunk: @@ -221,19 +221,36 @@ def call_runner(self, runner: BaseRunner) -> RunResult:
                     )
                 )
             status: Dict[str, str] = {}
+            result = None
+            thread_exception = None
             try:
                 result = runner.run_with_hooks(self.manifest)
-            except Exception as exc:
-                raise DbtInternalError(f"Unable to execute node: {exc}")
+            except Exception as e:
+                thread_exception = e
             finally:
                 finishctx = TimestampNamed("finished_at")
                 with finishctx, DbtModelState(status):
-                    fire_event(
-                        NodeFinished(
-                            node_info=runner.node.node_info,
-                            run_result=result.to_msg_dict(),
+                    if result is not None:
+                        fire_event(
+                            NodeFinished(
+                                node_info=runner.node.node_info,
+                                run_result=result.to_msg_dict(),
+                            )
                         )
-                    )
+                    else:
+                        msg = f"Exception on worker thread. {thread_exception}"
+                        fire_event(Note(msg=msg))
path: core/dbt/task/runnable.py
text: @user1:
Should we use the `GenericExceptionOnRun` event instead here?

@user2:
I'd vote for this