Repository: apache/buildstream
Issue: Failed builds not pushing artifacts on quit

[See original issue on GitLab](https://gitlab.com/BuildStream/buildstream/-/issues/534)
In GitLab by [[Gitlab user @knownexus]](https://gitlab.com/knownexus) on Aug 1, 2018, 10:12

## Summary

[//]: # (Summarize the bug encountered concisely)

If, when a build fails, the user quits instead of continuing, the artifact will not be pushed

## Steps to reproduce

[//]: # (How one can reproduce the issue - this is very important)
Build a project that fails,

When given the options of (t)terminate, (q)quit, (c)continue

select `q`

Check for the artifact in the cache

The artifact should be missing

## What is the expected correct behavior?

[//]: # (What you should see instead)
The failed build artifact should be cached even when the quit option is selected

## Possible fixes

[//]: # (If you can, link to the line of code that might be responsible for the problem)
One option is to have the artifact be cached at the point of the build failing, so at the point of choosing whether or not to continue, it is already cached

## Other relevant information

[//]: # (Add the BuildStream version. Substitute below the "x" by "1", "2" or the right version. Add more than one version if necessary)

* BuildStream version affected: /milestone %BuildStream_v1.4

----

Changed files:
src/buildstream/_scheduler/queues/queue.py, tests/integration/cachedfail.py, src/buildstream/_stream.py, src/buildstream/_scheduler/scheduler.py, tests/sourcecache/push.py, src/buildstream/_scheduler/queues/artifactpushqueue.py

diff --git a/src/buildstream/_scheduler/queues/artifactpushqueue.py b/src/buildstream/_scheduler/queues/artifactpushqueue.py
index 80cc81fae..95c444690 100644
--- a/src/buildstream/_scheduler/queues/artifactpushqueue.py
+++ b/src/buildstream/_scheduler/queues/artifactpushqueue.py
@@ -31,8 +31,8 @@ class ArtifactPushQueue(Queue):
     complete_name = "Artifacts Pushed"
     resources = [ResourceType.UPLOAD]
 
-    def __init__(self, scheduler, *, skip_uncached=False):
-        super().__init__(scheduler)
+    def __init__(self, scheduler, *, imperative=False, skip_uncached=False):
+        super().__init__(scheduler, imperative=imperative)
 
         self._skip_uncached = skip_uncached
 
diff --git a/src/buildstream/_scheduler/queues/queue.py b/src/buildstream/_scheduler/queues/queue.py
index 76688e75d..e9f934d57 100644
--- a/src/buildstream/_scheduler/queues/queue.py
+++ b/src/buildstream/_scheduler/queues/queue.py
@@ -64,7 +64,7 @@ class Queue:
     # Resources this queues' jobs want
     resources = []  # type: List[int]
 
-    def __init__(self, scheduler):
+    def __init__(self, scheduler, *, imperative=False):
 
         #
         # Private members
@@ -78,6 +78,11 @@ class Queue:
 
         self._required_element_check = False  # Whether we should check that elements are required before enqueuing
 
+        #
+        # Public members
+        #
+        self.imperative = imperative
+
         # Assert the subclass has setup class data
         assert self.action_name is not None
         assert self.complete_name is not None
diff --git a/src/buildstream/_scheduler/scheduler.py b/src/buildstream/_scheduler/scheduler.py
index 9d8e1c0f3..bba7de703 100644
--- a/src/buildstream/_scheduler/scheduler.py
+++ b/src/buildstream/_scheduler/scheduler.py
@@ -266,8 +266,8 @@ class Scheduler:
 
     # stop()
     #
-    # Stop queueing additional jobs, causes Scheduler.run()
-    # to return once all currently processing jobs are finished.
+    # Stop queueing additional imperative jobs, causes Scheduler.run()
+    # to return once all post-imperative jobs are finished.
     #
     def stop(self):
         self._queue_jobs = False
@@ -354,11 +354,27 @@ class Scheduler:
         ready = []
         process_queues = True
 
-        while self._queue_jobs and process_queues:
+        if self._queue_jobs:
+            # Scheduler.stop() was not called, consider all queues.
+            queues = self.queues
+        else:
+            # Limit processing to post-imperative queues
+            for queue in self.queues:
+                if queue.imperative:
+                    # Here the `queues` list will consists of the imperative queue along with
+                    # any subsequent queues, this means elements will be carried only from the
+                    # imerative queue onwards, and only post-imperative jobs will be processed.
+                    queues = self.queues[self.queues.index(queue) :]
+                    break
+            else:
+                # No imperative queue was marked, stop queueing any jobs
+                queues = []
+
+        while process_queues:
 
             # Pull elements forward through queues
             elements = []
-            for queue in self.queues:
+            for queue in queues:
                 queue.enqueue(elements)
                 elements = list(queue.dequeue())
 
@@ -373,13 +389,13 @@ class Scheduler:
             # to fetch tasks for elements which failed to pull, and
             # thus need all the pulls to complete before ever starting
             # a build
-            ready.extend(chain.from_iterable(q.harvest_jobs() for q in reversed(self.queues)))
+            ready.extend(chain.from_iterable(q.harvest_jobs() for q in reversed(queues)))
 
             # harvest_jobs() may have decided to skip some jobs, making
             # them eligible for promotion to the next queue as a side effect.
             #
             # If that happens, do another round.
-            process_queues = any(q.dequeue_ready() for q in self.queues)
+            process_queues = any(q.dequeue_ready() for q in queues)
 
         # Start the jobs
         #
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 428133839..5469b860f 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -423,7 +423,7 @@ class Stream:
 
         self._add_queue(FetchQueue(self._scheduler, skip_cached=True))
 
-        self._add_queue(BuildQueue(self._scheduler))
+        self._add_queue(BuildQueue(self._scheduler, imperative=True))
 
         if self._artifacts.has_push_remotes():
             self._add_queue(ArtifactPushQueue(self._scheduler, skip_uncached=True))
@@ -553,7 +553,7 @@ class Stream:
 
         self._add_queue(FetchQueue(self._scheduler))
 
-        self._add_queue(SourcePushQueue(self._scheduler))
+        self._add_queue(SourcePushQueue(self._scheduler, imperative=True))
 
         self._enqueue_plan(elements)
         self._run(announce_session=True)
@@ -652,7 +652,7 @@ class Stream:
 
         self._reset()
         self._add_queue(PullQueue(self._scheduler))
-        self._add_queue(ArtifactPushQueue(self._scheduler))
+        self._add_queue(ArtifactPushQueue(self._scheduler, imperative=True))
         self._enqueue_plan(elements)
         self._run(announce_session=True)
 
diff --git a/tests/integration/cachedfail.py b/tests/integration/cachedfail.py
index da1fff6c2..ed242c434 100644
--- a/tests/integration/cachedfail.py
+++ b/tests/integration/cachedfail.py
@@ -138,9 +138,6 @@ def test_build_depend_on_cached_fail(cli, datafiles):
 @pytest.mark.datafiles(DATA_DIR)
 @pytest.mark.parametrize("on_error", ("continue", "quit"))
 def test_push_cached_fail(cli, tmpdir, datafiles, on_error):
-    if on_error == "quit":
-        pytest.xfail("https://gitlab.com/BuildStream/buildstream/issues/534")
-
     project = str(datafiles)
     element_path = os.path.join(project, "elements", "element.bst")
 
@@ -185,9 +182,6 @@ def test_push_failed_missing_shell(cli, tmpdir, datafiles, on_error):
     When we don't have a valid shell, the artifact will be empty, not even the root directory.
     This ensures we handle the case of an entirely empty artifact correctly.
     """
-    if on_error == "quit":
-        pytest.xfail("https://gitlab.com/BuildStream/buildstream/issues/534")
-
     project = str(datafiles)
     element_path = os.path.join(project, "elements", "element.bst")
 
@@ -342,11 +336,6 @@ def test_nonstrict_retry_failed(cli, tmpdir, datafiles, use_share, retry):
         assert cli.get_element_state(project, "target.bst") == "failed"
 
         if use_share:
-            # Ensure that the target.bst has been shared, this is needed because of:
-            #     https://github.com/apache/buildstream/issues/534
-            result = cli.run(project=project, args=["artifact", "push", "target.bst"])
-            result.assert_success()
-
             # Delete the local cache, provoke pulling of the failed build
             cli.remove_artifact_from_cache(project, "target.bst")
 
diff --git a/tests/sourcecache/push.py b/tests/sourcecache/push.py
index 0931f684c..5fa936841 100644
--- a/tests/sourcecache/push.py
+++ b/tests/sourcecache/push.py
@@ -280,9 +280,8 @@ def test_source_push_build_fail(cli, tmpdir, datafiles):
         res.assert_main_error(ErrorDomain.STREAM, None)
         res.assert_task_error(ErrorDomain.ELEMENT, None)
 
-        # Sources are not pushed as the build queue is before the source push
-        # queue.
-        assert "Pushed source " not in res.stderr
+        # Sources are still pushed after failing a build
+        assert "Pushed source " in res.stderr
 
 
 # Test that source push succeeds if the source needs to be fetched