Repository: dbt-labs/dbt-core
Pull request: Add --fail-fast argument for dbt run and dbt test
resolves #1649

### Description

Optional flag `-x` or `--fail-fast` which enforce a `dbt run` to end up on first error (or test failure during `dbt test`). It's really beneficial during long-time runs and development (Continuous Integration pipelines will be grateful).

**I'm currently looking for suggestions how to (unit/integration) test the linker, graph parts and of course the tests severity levels.** 🍺 

### Checklist
 - [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
 - [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] I have updated the `CHANGELOG.md` and added information about my change to the "dbt next" section.


Problem statement:
Feature Request: --fail-fast argument for dbt run and dbt test
## Feature
`--fail-fast` argument for `dbt run` and `dbt test`

### Feature description
An optional argument, `--fail-fast`, for dbt run and dbt test which will stop model builds or test runs upon a single failure. It may also be worth adding a parameter which will define the amount of failures to allow before stopping.

### Who will this benefit?
All dbt users. Will be useful during the development process. Can also help reduce the amount of queries to the database.

### Questions
How should this work with test severity?

Hints from issue discussion:
Thanks @JasonGluck! I think we might have built this at a hackathon once, but I don't think the code ever made it into prod :/

I think the operative line of code is [right here](https://github.com/fishtown-analytics/dbt/blob/7177a6543b56d5d90446710e7295aaebeb415092/core/dbt/task/runnable.py#L113). I imagine we could add some logic there which consulted the value of the `--fail-fast` flag. If `--fail-fast` is provided, then dbt should "raise" regardless of the value supplied in `raise_on_first_error()`.

Is this something you'd be interested in contributing a PR for?
Dupe of #871 - closing that one.
This would save our team a lot of time!
I'm currently thinking about the behavior of `GraphRunnableTask.call_runner` in case of `dbt test` and modifying it with `fail-fast` based on discussion in #1675 (and setting `self._raise_next_tick` based on `result.fail`). Let me know if I'm wrong with that but it seems that during testing the `job_queue` structure is flat (as expected) with no dependents between nodes. But if we're waiting for `self._raise_next_tick` to be considered (throw `_raise_set_error()`) it will be [called only during submitting tasks to threads pool](https://github.com/fishtown-analytics/dbt/blob/b77b3a45661e68a647a6be91e197c1cd13545982/core/dbt/task/runnable.py#L211) (before of tests execution) and [after the end of all tasks (tests) execution in](https://github.com/fishtown-analytics/dbt/blob/b77b3a45661e68a647a6be91e197c1cd13545982/core/dbt/task/runnable.py#L224). So we currently changing nothing than disabling summary list after the execution.

Now I'm looking for place where we can gentle raise in-between testing. Of course I really appreciate any help with that 🍺 

My current state of work could be found at: [raalsky/fail-fast](https://github.com/fishtown-analytics/dbt/compare/dev/octavius-catto...Raalsky:fail-fast)

I'm also thinking about testing these parts of codebase in some sort of providing a desired ordered list of nodes to `job_queue` (which is really brutal in some sense 🌵 ).

Of course If @sjwhitworth is still working on it just let me know 😉 
hey @Raalsky - this is a really good point! Tagging @beckjake to advise -- I'm definitely seeing the behavior you're describing when I run your branch locally :)

Jake, what's the best way to bubble up test failures to fail-fast? Or, should we reconsider how we feed test nodes into the runner?
What a great point! That's definitely a hole in the process. I think the key issue is using `job_queue.join()` which ultimately does this on a `queue.PriorityQueue`:
```
with self.all_tasks_done:
    while self.unfinished_tasks:
        self.all_tasks_done.wait()
```

We could do this (but probably abstract it so we aren't reaching so far into internals):
```
with self.job_queue.inner.all_tasks_done:
    while self.job_queue.inner.unfinished_tasks:
        self._raise_next_tick()
        self.job_queue.inner.all_tasks_done.wait()
```


I would probably do that by writing a `join_with(self, func)` method inside the graph queue code and calling it like `self.job_queue.join_with(self._raise_next_tick)`. It's going to be a bit messy no matter what, I suspect.
Hi @drewbanin and @beckjake
I much appreciate your suggestions 🚀 . I'll be back in a few days with updates.
@beckjake 
Based on implementation of `PriorityQueue` (`Queue`) I think the reformat of `job_queue.join()` might be insufficient. In the following:
```python
with self.job_queue.inner.all_tasks_done:
    while self.job_queue.inner.unfinished_tasks:
        tell_the_world_is_ending()
        self.job_queue.inner.all_tasks_done.wait()
```
I think the `tell_the_world_is_ending()` will be called only once during `dbt test`. The reason is located in implementation of `job_queue.task_done()` which is called at the end of execution of every task (test):
```python
    def task_done(self):
        '''Indicate that a formerly enqueued task is complete.

        Used by Queue consumer threads.  For each get() used to fetch a task,
        a subsequent call to task_done() tells the queue that the processing
        on the task is complete.

        If a join() is currently blocking, it will resume when all items
        have been processed (meaning that a task_done() call was received
        for every item that had been put() into the queue).

        Raises a ValueError if called more times than there were items
        placed in the queue.
        '''
        with self.all_tasks_done:
            unfinished = self.unfinished_tasks - 1
            if unfinished <= 0:
                if unfinished < 0:
                    raise ValueError('task_done() called too many times')
                self.all_tasks_done.notify_all()
            self.unfinished_tasks = unfinished
```
We're blocking on `self.job_queue.inner.all_tasks_done.wait()` which will be called by `task_done()` throw `self.all_tasks_done.notify_all()` (as it's simple Mutex) only when there is noting to process actually (so basically after the end of all tests execution).

I think we should consider also rewrite of `task_done()` with some mechanism of `notify_after_something_was_done` in addition to `.join()`.
OR
Mixing the task submitting part with some sort of wait (maybe based on number of threads). Of course it's only direction:
```python
        while not nothing_to_be_processed:
            node = self.job_queue.get()
            self._raise_set_error()
            runner = self.get_runner(node)
            ...
            self._submit(pool, args, callback)
            # Only an approximation of queue size!
            if self.job_queue.qsize() >= self.config.threads:
                 self.job_queue.wait_until_something_was_done()
```
Keep in mind it's not only about `dbt test` but also in `dbt run` when we have a bunch of independent tasks (aka. large "layer") for example transforming sources etc.
I've pushed an MVP to [Raalsky/fail-fast](https://github.com/fishtown-analytics/dbt/compare/dev/octavius-catto...Raalsky:fail-fast) (need a lot in carrying about concurrency 😭 ) but it works pretty close to what I've expected.

Commit message: Added --fail-fast argument for dbt run and dbt test

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9e1d53bda28..e8317f9dc1e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,8 +1,13 @@
 ## dbt next (release TBD)
+### Features
+- Added --fail-fast argument for dbt run and dbt test to fail on first test failure or runtime error. ([#1649](https://github.com/fishtown-analytics/dbt/issues/1649))
 
 ### Fixes
 - When a jinja value is undefined, give a helpful error instead of failing with cryptic "cannot pickle ParserMacroCapture" errors ([#2110](https://github.com/fishtown-analytics/dbt/issues/2110), [#2184](https://github.com/fishtown-analytics/dbt/pull/2184))
 
+Contributors:
+ - [@raalsky](https://github.com/Raalsky) ([#2224](https://github.com/fishtown-analytics/dbt/pull/2224))
+
 ## dbt 0.16.0 (Release date TBD)
 
 ### Fixes
diff --git a/core/dbt/exceptions.py b/core/dbt/exceptions.py
index cbbe3c74752..894d50758b9 100644
--- a/core/dbt/exceptions.py
+++ b/core/dbt/exceptions.py
@@ -281,6 +281,17 @@ def __init__(self, message, project=None, result_type='invalid_project'):
         self.result_type = result_type
 
 
+class FailFastException(Exception):
+    CODE = 10013
+    MESSAGE = 'FailFast Error'
+
+    def __init__(self, result=None):
+        self.result = result
+
+    def __str__(self):
+        return 'Falling early due to test failure or runtime error'
+
+
 class DbtProjectError(DbtConfigError):
     pass
 
diff --git a/core/dbt/linker.py b/core/dbt/linker.py
index 991d704901d..1170806b11b 100644
--- a/core/dbt/linker.py
+++ b/core/dbt/linker.py
@@ -48,6 +48,8 @@ def __init__(self, graph, manifest):
         self._scores = self._calculate_scores()
         # populate the initial queue
         self._find_new_additions()
+        # awaits after task end
+        self.some_task_done = threading.Condition(self.lock)
 
     def _include_in_cost(self, node_id):
         node = self.manifest.expect(node_id)
@@ -153,6 +155,7 @@ def mark_done(self, node_id):
             self.graph.remove_node(node_id)
             self._find_new_additions()
             self.inner.task_done()
+            self.some_task_done.notify_all()
 
     def _mark_in_progress(self, node_id):
         """Mark the node as 'in progress'.
@@ -171,6 +174,11 @@ def join(self):
         """
         self.inner.join()
 
+    def wait_until_something_was_done(self):
+        with self.some_task_done:
+            self.some_task_done.wait()
+            return self.inner.unfinished_tasks
+
 
 class Linker:
     def __init__(self, data=None):
diff --git a/core/dbt/main.py b/core/dbt/main.py
index 12fa3bb2301..0da30df2695 100644
--- a/core/dbt/main.py
+++ b/core/dbt/main.py
@@ -398,6 +398,14 @@ def _build_run_subparser(subparsers, base_subparser):
         help='''
         Compile SQL and execute against the current target database.
         ''')
+    run_sub.add_argument(
+        '-x',
+        '--fail-fast',
+        action='store_true',
+        help='''
+            Stop execution upon a first failure.
+            '''
+    )
     run_sub.set_defaults(cls=run_task.RunTask, which='run', rpc_method='run')
     return run_sub
 
@@ -555,6 +563,14 @@ def _build_test_subparser(subparsers, base_subparser):
         Run constraint validations from schema.yml files
         '''
     )
+    sub.add_argument(
+        '-x',
+        '--fail-fast',
+        action='store_true',
+        help='''
+        Stop execution upon a first test failure.
+        '''
+    )
 
     sub.set_defaults(cls=test_task.TestTask, which='test', rpc_method='test')
     return sub
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index 1aed744c390..09dc0588b6b 100644
--- a/core/dbt/task/runnable.py
+++ b/core/dbt/task/runnable.py
@@ -25,7 +25,10 @@
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.results import ExecutionResult
 from dbt.exceptions import (
-    InternalException, NotImplementedException, RuntimeException
+    InternalException,
+    NotImplementedException,
+    RuntimeException,
+    FailFastException
 )
 from dbt.linker import Linker, GraphQueue
 from dbt.perf_utils import get_full_manifest
@@ -167,11 +170,16 @@ def call_runner(self, runner):
                 with finishctx, DbtModelState(status):
                     logger.debug('Finished running node {}'.format(
                         runner.node.unique_id))
-        if result.error is not None and self.raise_on_first_error():
+
+        fail_fast = getattr(self.config.args, 'fail_fast', False)
+
+        if (result.fail is not None or result.error is not None) and fail_fast:
+            self._raise_next_tick = FailFastException(result)
+        elif result.error is not None and self.raise_on_first_error():
             # if we raise inside a thread, it'll just get silently swallowed.
             # stash the error message we want here, and it will check the
             # next 'tick' - should be soon since our thread is about to finish!
-            self._raise_next_tick = result.error
+            self._raise_next_tick = RuntimeException(result.error)
 
         return result
 
@@ -191,7 +199,7 @@ def _submit(self, pool, args, callback):
 
     def _raise_set_error(self):
         if self._raise_next_tick is not None:
-            raise RuntimeException(self._raise_next_tick)
+            raise self._raise_next_tick
 
     def run_queue(self, pool):
         """Given a pool, submit jobs from the queue to the pool.
@@ -226,7 +234,15 @@ def callback(result):
             self._submit(pool, args, callback)
 
         # block on completion
-        self.job_queue.join()
+        if getattr(self.config.args, 'fail_fast', False):
+            # checkout for an errors after task completion in case of
+            # fast failure
+            while self.job_queue.wait_until_something_was_done():
+                self._raise_set_error()
+        else:
+            # wait until every task will be complete
+            self.job_queue.join()
+
         # if an error got set during join(), raise it.
         self._raise_set_error()
 
@@ -255,6 +271,34 @@ def _handle_result(self, result):
                 cause = None
             self._mark_dependent_errors(node.unique_id, result, cause)
 
+    def _cancel_connections(self, pool):
+        """Given a pool, cancel all adapter connections and wait until all
+        runners gentle terminates.
+        """
+        pool.close()
+        pool.terminate()
+
+        adapter = get_adapter(self.config)
+
+        if not adapter.is_cancelable():
+            msg = ("The {} adapter does not support query "
+                   "cancellation. Some queries may still be "
+                   "running!".format(adapter.type()))
+
+            yellow = dbt.ui.printer.COLOR_FG_YELLOW
+            dbt.ui.printer.print_timestamped_line(msg, yellow)
+            raise
+
+        for conn_name in adapter.cancel_open_connections():
+            if self.manifest is not None:
+                node = self.manifest.nodes.get(conn_name)
+                if node is not None and node.is_ephemeral_model:
+                    continue
+            # if we don't have a manifest/don't have a node, print anyway.
+            dbt.ui.printer.print_cancel_line(conn_name)
+
+        pool.join()
+
     def execute_nodes(self):
         num_threads = self.config.threads
         target_name = self.config.target_name
@@ -270,34 +314,15 @@ def execute_nodes(self):
         try:
             self.run_queue(pool)
 
-        except KeyboardInterrupt:
-            pool.close()
-            pool.terminate()
-
-            adapter = get_adapter(self.config)
-
-            if not adapter.is_cancelable():
-                msg = ("The {} adapter does not support query "
-                       "cancellation. Some queries may still be "
-                       "running!".format(adapter.type()))
-
-                yellow = dbt.ui.printer.COLOR_FG_YELLOW
-                dbt.ui.printer.print_timestamped_line(msg, yellow)
-                raise
-
-            for conn_name in adapter.cancel_open_connections():
-                if self.manifest is not None:
-                    node = self.manifest.nodes.get(conn_name)
-                    if node is not None and node.is_ephemeral_model:
-                        continue
-                # if we don't have a manifest/don't have a node, print anyway.
-                dbt.ui.printer.print_cancel_line(conn_name)
-
-            pool.join()
+        except FailFastException as failure:
+            self._cancel_connections(pool)
+            dbt.ui.printer.print_run_result_error(failure.result)
+            raise
 
+        except KeyboardInterrupt:
+            self._cancel_connections(pool)
             dbt.ui.printer.print_run_end_messages(self.node_results,
-                                                  early_exit=True)
-
+                                                  keyboard_interrupt=True)
             raise
 
         pool.close()
diff --git a/core/dbt/ui/printer.py b/core/dbt/ui/printer.py
index c913274b611..1c6ff051c79 100644
--- a/core/dbt/ui/printer.py
+++ b/core/dbt/ui/printer.py
@@ -348,11 +348,11 @@ def print_skip_caused_by_error(
 
 
 def print_end_of_run_summary(
-    num_errors: int, num_warnings: int, early_exit: bool = False
+    num_errors: int, num_warnings: int, keyboard_interrupt: bool = False
 ) -> None:
     error_plural = dbt.utils.pluralize(num_errors, 'error')
     warn_plural = dbt.utils.pluralize(num_warnings, 'warning')
-    if early_exit:
+    if keyboard_interrupt:
         message = yellow('Exited because of keyboard interrupt.')
     elif num_errors > 0:
         message = red("Completed with {} and {}:".format(
@@ -367,11 +367,13 @@ def print_end_of_run_summary(
     logger.info('{}'.format(message))
 
 
-def print_run_end_messages(results, early_exit: bool = False) -> None:
+def print_run_end_messages(results, keyboard_interrupt: bool = False) -> None:
     errors = [r for r in results if r.error is not None or r.fail]
     warnings = [r for r in results if r.warn]
     with DbtStatusMessage(), InvocationProcessor():
-        print_end_of_run_summary(len(errors), len(warnings), early_exit)
+        print_end_of_run_summary(len(errors),
+                                 len(warnings),
+                                 keyboard_interrupt)
 
         for error in errors:
             print_run_result_error(error, is_warning=False)


Code review comments:
diff_hunk: @@ -281,6 +281,17 @@ def __init__(self, message, project=None, result_type='invalid_project'):
         self.result_type = result_type
 
 
+class FailFastException(Exception):
path: core/dbt/exceptions.py
text: @author:
@user1 What about this part? I'm not feeling 100% ok with all this logic about `Exception/Errors` and I was working about something similar to the other ones but without deep understanding what should be here.

@user1:
Hmm, that's an interesting question. Maybe derive from `RuntimeException` and skip `__init__`/`__str__`? It would be nice to attach the failed node to the exception, if we can!

@user1:
So when raising it, something like this (if it works - I have not tested literally anything about it!):
`raise FailFastException(result.error, node=getattr(result, 'node', None))`

@author:
Ok. Thanks

@author:
Done 😉