_stream.py: Pull missing artifacts in push()juerg/push

As per #819, BuildStream should pull missing artifacts by default. The
previous behavior was to only pull missing buildtrees. A top-level
`--no-pull` option can easily be supported in the future.

This change makes it possible to use a single scheduler session (with
concurrent pull and push jobs). This commit also simplifies the code as
it removes the `sched_error_action` emulation, using the regular
frontend code path instead.

2 files changed, 8 insertions, 46 deletions

diff --git a/src/buildstream/_scheduler/queues/artifactpushqueue.py b/src/buildstream/_scheduler/queues/artifactpushqueue.py
index 6ed79a5ba..79597c704 100644
--- a/src/buildstream/_scheduler/queues/artifactpushqueue.py
+++ b/src/buildstream/_scheduler/queues/artifactpushqueue.py
@@ -32,7 +32,7 @@ class ArtifactPushQueue(Queue):
     complete_name = "Artifacts Pushed"
     resources = [ResourceType.UPLOAD]
 
-    def __init__(self, scheduler, *, skip_uncached=True):
+    def __init__(self, scheduler, *, skip_uncached=False):
         super().__init__(scheduler)
 
         self._skip_uncached = skip_uncached
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index fbb2fca69..3b0a308e7 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -48,7 +48,7 @@ from .element import Element
 from ._pipeline import Pipeline
 from ._profile import Topics, PROFILER
 from ._state import State
-from .types import _KeyStrength, _PipelineSelection, _SchedulerErrorAction, _Scope
+from .types import _KeyStrength, _PipelineSelection, _Scope
 from .plugin import Plugin
 from . import utils, _yaml, _site
 
@@ -301,7 +301,7 @@ class Stream:
         self._add_queue(BuildQueue(self._scheduler))
 
         if self._artifacts.has_push_remotes():
-            self._add_queue(ArtifactPushQueue(self._scheduler))
+            self._add_queue(ArtifactPushQueue(self._scheduler, skip_uncached=True))
 
         if self._sourcecache.has_push_remotes():
             self._add_queue(SourcePushQueue(self._scheduler))
@@ -484,49 +484,11 @@ class Stream:
 
         self._pipeline.assert_consistent(elements)
 
-        # Check if we require a pull queue, with given artifact state and context
-        require_buildtrees = self._buildtree_pull_required(elements)
-        if require_buildtrees:
-            self._message(MessageType.INFO, "Attempting to fetch missing artifact buildtrees")
-            self._add_queue(PullQueue(self._scheduler))
-            self._enqueue_plan(require_buildtrees)
-
-        # Before we try to push the artifacts, ensure they're cached
-        cached_elements = []
-        uncached_elements = []
-        self._message(MessageType.INFO, "Verifying that elements are cached")
-        for element in elements:
-            if element._cached():
-                cached_elements.append(element)
-            else:
-                msg = "{} is not cached".format(element.name)
-                if self._context.sched_error_action != _SchedulerErrorAction.CONTINUE:
-                    raise StreamError("Push failed: " + msg)
-
-                self._message(MessageType.WARN, msg)
-                uncached_elements.append(element)
-
-        if cached_elements:
-            self._scheduler.clear_queues()
-            push_queue = ArtifactPushQueue(self._scheduler)
-            self._add_queue(push_queue)
-            self._enqueue_plan(cached_elements, queue=push_queue)
-            self._run(announce_session=True)
-
-        # If the user has selected to continue on error, fail the command
-        # and print a summary of artifacts which could not be pushed
-        #
-        # NOTE: Usually we check the _SchedulerErrorAction when a *job* has failed.
-        #       However, we cannot create a PushQueue job unless we intentionally
-        #       ready an uncached element in the PushQueue.
-        if self._context.sched_error_action == _SchedulerErrorAction.CONTINUE and uncached_elements:
-            names = [element.name for element in uncached_elements]
-            fail_str = (
-                "Error while pushing. The following elements were not pushed as they are "
-                "not yet cached:\n\n\t{}\n".format("\n\t".join(names))
-            )
-
-            raise StreamError(fail_str)
+        self._scheduler.clear_queues()
+        self._add_queue(PullQueue(self._scheduler))
+        self._add_queue(ArtifactPushQueue(self._scheduler))
+        self._enqueue_plan(elements)
+        self._run(announce_session=True)
 
     # checkout()
     #