Repository: dbt-labs/dbt-core
Issue: [Grace Kelly] Exceptions in `call_runner` cause thread pool to hang

## Issue

### Issue description
With dbt's new approach to multithreading, exceptions raised in `runner.call_runner` cause the thread pool to hang. I'm having a hard time finding the expected behavior of a thread pool when a worker raises an exception, but it appears that 1) the exception message is swallowed and 2) the pool ends up blocking on the next `get()` call.

I can think of two ways to work around this:
1. methods called by `call_runner` should _never_ raise
2. a wrapper function around `call_runner` should be provided which logs exceptions but does not raise them

This is the general idea behind the bug, but the specific situation where is related to the [raise_on_first_error](https://github.com/fishtown-analytics/dbt/blob/3c25a9b40d725f35ea8a0ef78b61c511113596c4/dbt/runner.py#L84) attribute of certain node runners. The `CompileRunner` has this value set to `True`, so exceptions during compilation cause an exception to be raise in the `call_runner` method. No other runners have `raise_on_first_error` enabled, so I'm only able to reproduce this bug with `dbt compile` (and `dbt docs generate`, which invokes `dbt compile`).

In addition to ensuring that `call_runner` never raises (or handling the exceptions in the pool, if that's possible), we should do something about `raise_on_first_error`. I'm unsure how useful this attribute is in its current incarnation, and I think I'd be ok with just removing it. The other option is to keep that attribute, and then flip some flag that ultimately terminates the pool from the main thread.

### Results

The thread pool blocked on a `get()` call after an exception was raised in `dbt compile`.

### System information
```
dbt version: 0.12.2-rc1
python version: 3.7.1
python path: /Users/drew/fishtown/dbt/env/bin/python3
os info: Darwin-18.2.0-x86_64-i386-64bit
```

### Steps to reproduce
This bug manifests when a CompilationError exists at "compile time". Crucially though, most compile-time errors are also parse-time errors. Here's an example of a model that will succeed at parse-time (allowing dbt to proceed to compilation), but that will fail at compile time;

```
-- models/my_model.sql
{% if execute %}
    select * from {{ ref('notfound') }}
{% else %}
    select 1 as id
{% endif %}
```

Then run:
```
$ dbt compile
```

Changed files:
dbt/runner.py

diff --git a/dbt/runner.py b/dbt/runner.py
index 8b6e2f87b..3d8e15840 100644
--- a/dbt/runner.py
+++ b/dbt/runner.py
@@ -54,6 +54,7 @@ class RunManager(object):
         ])
         self.node_results = []
         self._skipped_children = {}
+        self._raise_next_tick = None
 
     def get_runner(self, node):
         adapter = get_adapter(self.config)
@@ -82,7 +83,11 @@ class RunManager(object):
             runner.after_execute(result)
 
         if result.errored and runner.raise_on_first_error():
-            raise dbt.exceptions.RuntimeException(result.error)
+            # if we raise inside a thread, it'll just get silently swallowed.
+            # stash the error message we want on the RunManager, and it will
+            # check the next 'tick' - should be soon since our thread is about
+            # to finish!
+            self._raise_next_tick = result.error
 
         return result
 
@@ -100,16 +105,23 @@ class RunManager(object):
         else:
             pool.apply_async(self.call_runner, args=args, callback=callback)
 
+    def _raise_set_error(self):
+        if self._raise_next_tick is not None:
+            raise dbt.exceptions.RuntimeException(self._raise_next_tick)
+
     def run_queue(self, pool):
         """Given a pool, submit jobs from the queue to the pool.
         """
         def callback(result):
-            """A callback to handle results."""
+            """Note: mark_done, at a minimum, must happen here or dbt will
+            deadlock during ephemeral result error handling!
+            """
             self._handle_result(result)
             self.job_queue.mark_done(result.node.unique_id)
 
         while not self.job_queue.empty():
             node = self.job_queue.get()
+            self._raise_set_error()
             runner = self.get_runner(node)
             # we finally know what we're running! Make sure we haven't decided
             # to skip it due to upstream failures
@@ -118,14 +130,18 @@ class RunManager(object):
                 runner.do_skip(cause=cause)
             args = (runner,)
             self._submit(pool, args, callback)
+
         # block on completion
         self.job_queue.join()
+        # if an error got set during join(), raise it.
+        self._raise_set_error()
 
         return
 
     def _handle_result(self, result):
-        """Note: this happens inside an apply_async() callback, so it must be
-        "fast". (The pool worker thread will block!)
+        """Mark the result as completed, insert the `CompiledResultNode` into
+        the manifest, and mark any descendants (potentially with a 'cause' if
+        the result was an ephemeral model) as skipped.
         """
         is_ephemeral = self.Runner.is_ephemeral_model(result.node)
         if not is_ephemeral: