WIP: Perform artifact cache expiry135-expire-artifacts-in-local-cache-clean

19 files changed, 1001 insertions, 41 deletions

diff --git a/NEWS b/NEWS
index 2d0c63406..690fdb301 100644
--- a/NEWS
+++ b/NEWS
@@ -16,6 +16,9 @@ buildstream 1.1.3
 
   o BuildStream now requires python version >= 3.5
   o Added new `bst init` command to initialize a new project.
+  o BuildStream will now automatically clean up old artifacts when it
+    runs out of space. The exact behavior is configurable in
+    `userconf.yaml`.
 
   o Cross junction tracking is now disabled by default for projects
     which can support this by using project.refs ref-storage
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index 2d745f8c2..bf8ff4ae5 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -21,7 +21,8 @@ import os
 import string
 from collections import Mapping, namedtuple
 
-from .._exceptions import ImplError, LoadError, LoadErrorReason
+from ..element import _KeyStrength
+from .._exceptions import ArtifactError, ImplError, LoadError, LoadErrorReason
 from .._message import Message, MessageType
 from .. import utils
 from .. import _yaml
@@ -61,7 +62,11 @@ class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push')):
 class ArtifactCache():
     def __init__(self, context):
         self.context = context
+        self.required_artifacts = set()
         self.extractdir = os.path.join(context.artifactdir, 'extract')
+        self.max_size = context.cache_quota
+        self.estimated_size = None
+
         self.global_remote_specs = []
         self.project_remote_specs = {}
 
@@ -162,6 +167,102 @@ class ArtifactCache():
                                   (str(provenance)))
         return cache_specs
 
+    # append_required_artifacts():
+    #
+    # Append to the list of elements whose artifacts are required for
+    # the current run. Artifacts whose elements are in this list will
+    # be locked by the artifact cache and not touched for the duration
+    # of the current pipeline.
+    #
+    # Args:
+    #     elements (iterable): A set of elements to mark as required
+    #
+    def append_required_artifacts(self, elements):
+        # We lock both strong and weak keys - deleting one but not the
+        # other won't save space in most cases anyway, but would be a
+        # user inconvenience.
+
+        for element in elements:
+            strong_key = element._get_cache_key(strength=_KeyStrength.STRONG)
+            weak_key = element._get_cache_key(strength=_KeyStrength.WEAK)
+
+            for key in (strong_key, weak_key):
+                if key and key not in self.required_artifacts:
+                    self.required_artifacts.add(key)
+
+                    # We also update the usage times of any artifacts
+                    # we will be using, which helps preventing a
+                    # buildstream process that runs in parallel with
+                    # this one from removing artifacts in-use.
+                    try:
+                        self.update_atime(element, key)
+                    except FileNotFoundError:
+                        pass
+
+    # clean():
+    #
+    # Clean the artifact cache as much as possible.
+    #
+    def clean(self):
+        artifacts = self.list_artifacts()
+
+        while self.calculate_cache_size() >= self.context.cache_quota - self.context.cache_lower_threshold:
+            try:
+                to_remove = artifacts.pop(0)
+            except IndexError:
+                # If too many artifacts are required, and we therefore
+                # can't remove them, we have to abort the build.
+                #
+                # FIXME: Asking the user what to do may be neater
+                default_conf = os.path.join(os.environ['XDG_CONFIG_HOME'],
+                                            'buildstream.conf')
+                detail = ("There is not enough space to build the given element.\n"
+                          "Please increase the cache-quota in {}."
+                          .format(self.context.config_origin or default_conf))
+
+                if self.calculate_cache_size() > self.context.cache_quota:
+                    raise ArtifactError("Cache too full. Aborting.",
+                                        detail=detail,
+                                        reason="cache-too-full")
+                else:
+                    break
+
+            key = to_remove.rpartition('/')[2]
+            if key not in self.required_artifacts:
+                self.remove(to_remove)
+
+        # This should be O(1) if implemented correctly
+        return self.calculate_cache_size()
+
+    # get_approximate_cache_size()
+    #
+    # A cheap method that aims to serve as an upper limit on the
+    # artifact cache size.
+    #
+    # The cache size reported by this function will normally be larger
+    # than the real cache size, since it is calculated using the
+    # pre-commit artifact size, but for very small artifacts in
+    # certain caches additional overhead could cause this to be
+    # smaller than, but close to, the actual size.
+    #
+    # Nonetheless, in practice this should be safe to use as an upper
+    # limit on the cache size.
+    #
+    # If the cache has built-in constant-time size reporting, please
+    # feel free to override this method with a more accurate
+    # implementation.
+    #
+    # Returns:
+    #     (int) An approximation of the artifact cache size.
+    #
+    def get_approximate_cache_size(self):
+        # If we don't currently have an estimate, figure out the real
+        # cache size.
+        if self.estimated_size is None:
+            self.estimated_size = self.calculate_cache_size()
+
+        return self.estimated_size
+
     ################################################
     # Abstract methods for subclasses to implement #
     ################################################
@@ -191,6 +292,44 @@ class ArtifactCache():
         raise ImplError("Cache '{kind}' does not implement contains()"
                         .format(kind=type(self).__name__))
 
+    # list_artifacts():
+    #
+    # List artifacts in this cache in LRU order.
+    #
+    # Returns:
+    #     ([str]) - A list of artifact names as generated by
+    #               `ArtifactCache.get_artifact_fullname` in LRU order
+    #
+    def list_artifacts(self):
+        raise ImplError("Cache '{kind}' does not implement list_artifacts()"
+                        .format(kind=type(self).__name__))
+
+    # remove():
+    #
+    # Removes the artifact for the specified ref from the local
+    # artifact cache.
+    #
+    # Args:
+    #     ref (artifact_name): The name of the artifact to remove (as
+    #                          generated by
+    #                          `ArtifactCache.get_artifact_fullname`)
+    #
+    def remove(self, artifact_name):
+        raise ImplError("Cache '{kind}' does not implement remove()"
+                        .format(kind=type(self).__name__))
+
+    # update_atime():
+    #
+    # Update the access time of an element.
+    #
+    # Args:
+    #     element (Element): The Element to mark
+    #     key (str): The cache key to use
+    #
+    def update_atime(self, element, key):
+        raise ImplError("Cache '{kind}' does not implement update_atime()"
+                        .format(kind=type(self).__name__))
+
     # extract():
     #
     # Extract cached artifact for the specified Element if it hasn't
@@ -320,6 +459,20 @@ class ArtifactCache():
         raise ImplError("Cache '{kind}' does not implement link_key()"
                         .format(kind=type(self).__name__))
 
+    # calculate_cache_size()
+    #
+    # Return the real artifact cache size.
+    #
+    # Implementations should also use this to update estimated_size.
+    #
+    # Returns:
+    #
+    # (int) The size of the artifact cache.
+    #
+    def calculate_cache_size(self):
+        raise ImplError("Cache '{kind}' does not implement calculate_cache_size()"
+                        .format(kind=type(self).__name__))
+
     ################################################
     #               Local Private Methods          #
     ################################################
@@ -361,6 +514,30 @@ class ArtifactCache():
         with self.context.timed_activity("Initializing remote caches", silent_nested=True):
             self.initialize_remotes(on_failure=remote_failed)
 
+    # _add_artifact_size()
+    #
+    # Since we cannot keep track of the cache size between threads,
+    # this method will be called by the main process every time a
+    # process that added something to the cache finishes.
+    #
+    # This will then add the reported size to
+    # ArtifactCache.estimated_size.
+    #
+    def _add_artifact_size(self, artifact_size):
+        if not self.estimated_size:
+            self.estimated_size = self.calculate_cache_size()
+
+        self.estimated_size += artifact_size
+
+    # _set_cache_size()
+    #
+    # Similarly to the above method, when we calculate the actual size
+    # in a child thread, we can't update it. We instead pass the value
+    # back to the main thread and update it there.
+    #
+    def _set_cache_size(self, cache_size):
+        self.estimated_size = cache_size
+
 
 # _configured_remote_artifact_cache_specs():
 #
diff --git a/buildstream/_artifactcache/ostreecache.py b/buildstream/_artifactcache/ostreecache.py
index 707a974eb..bb8f4fce8 100644
--- a/buildstream/_artifactcache/ostreecache.py
+++ b/buildstream/_artifactcache/ostreecache.py
@@ -59,6 +59,9 @@ class OSTreeCache(ArtifactCache):
         self._has_fetch_remotes = False
         self._has_push_remotes = False
 
+        # A cached artifact cache size (irony?)
+        self.cache_size = None
+
     ################################################
     #     Implementation of abstract methods       #
     ################################################
@@ -90,6 +93,21 @@ class OSTreeCache(ArtifactCache):
         ref = self.get_artifact_fullname(element, key)
         return _ostree.exists(self.repo, ref)
 
+    def list_artifacts(self):
+        return _ostree.list_artifacts(self.repo)
+
+    def remove(self, artifact_name):
+        # We cannot defer pruning, unfortunately, because we could
+        # otherwise not figure out how much space was freed by the
+        # removal, and would therefore not be able to expire the
+        # correct number of artifacts.
+        self.cache_size -= _ostree.remove(self.repo, artifact_name, defer_prune=False)
+
+    def update_atime(self, element, key):
+        ref = self.get_artifact_fullname(element, key)
+        ref_file = os.path.join(self.repo.get_path().get_path(), 'refs', 'heads', ref)
+        os.utime(ref_file)
+
     def extract(self, element, key):
         ref = self.get_artifact_fullname(element, key)
 
@@ -99,6 +117,9 @@ class OSTreeCache(ArtifactCache):
         # Extracting a nonexistent artifact is a bug
         assert rev, "Artifact missing for {}".format(ref)
 
+        ref_file = os.path.join(self.repo.get_path().get_path(), 'refs', 'heads', ref)
+        os.utime(ref_file)
+
         dest = os.path.join(self.extractdir, element._get_project().name, element.normal_name, rev)
         if os.path.isdir(dest):
             # artifact has already been extracted
@@ -134,6 +155,10 @@ class OSTreeCache(ArtifactCache):
         except OSTreeError as e:
             raise ArtifactError("Failed to commit artifact: {}".format(e)) from e
 
+        self.append_required_artifacts([element])
+
+        self.cache_size = None
+
     def can_diff(self):
         return True
 
@@ -167,6 +192,9 @@ class OSTreeCache(ArtifactCache):
                 # fetch the artifact from highest priority remote using the specified cache key
                 remote_name = self._ensure_remote(self.repo, remote.pull_url)
                 _ostree.fetch(self.repo, remote=remote_name, ref=ref, progress=progress)
+
+                self.append_required_artifacts([element])
+
                 return True
             except OSTreeError:
                 # Try next remote
@@ -201,6 +229,13 @@ class OSTreeCache(ArtifactCache):
 
         return any_pushed
 
+    def calculate_cache_size(self):
+        if self.cache_size is None:
+            self.cache_size = utils._get_dir_size(self.repo.get_path().get_path())
+            self.estimated_size = self.cache_size
+
+        return self.cache_size
+
     def initialize_remotes(self, *, on_failure=None):
         remote_specs = self.global_remote_specs.copy()
 
diff --git a/buildstream/_artifactcache/tarcache.py b/buildstream/_artifactcache/tarcache.py
index ab814abb0..4e9f5f9fa 100644
--- a/buildstream/_artifactcache/tarcache.py
+++ b/buildstream/_artifactcache/tarcache.py
@@ -36,6 +36,7 @@ class TarCache(ArtifactCache):
 
         self.tardir = os.path.join(context.artifactdir, 'tar')
         os.makedirs(self.tardir, exist_ok=True)
+        self.cache_size = None
 
     ################################################
     #     Implementation of abstract methods       #
@@ -44,6 +45,34 @@ class TarCache(ArtifactCache):
         path = os.path.join(self.tardir, _tarpath(element, key))
         return os.path.isfile(path)
 
+    # list_artifacts():
+    #
+    # List artifacts in this cache in LRU order.
+    #
+    # Returns:
+    #     (list) - A list of refs in LRU order
+    #
+    def list_artifacts(self):
+        artifacts = list(utils.list_relative_paths(self.tardir, list_dirs=False))
+        mtimes = [os.path.getmtime(os.path.join(self.tardir, artifact))
+                  for artifact in artifacts if artifact]
+
+        # We need to get rid of the tarfile extension to get a proper
+        # ref - os.splitext doesn't do this properly, unfortunately.
+        artifacts = [artifact[:-len('.tar.bz2')] for artifact in artifacts]
+
+        return [name for _, name in sorted(zip(mtimes, artifacts))]
+
+    # remove()
+    #
+    # Implements artifactcache.remove().
+    #
+    def remove(self, artifact_name):
+        artifact = os.path.join(self.tardir, artifact_name + '.tar.bz2')
+        size = os.stat(artifact, follow_symlinks=False).st_size
+        os.remove(artifact)
+        self.cache_size -= size
+
     def commit(self, element, content, keys):
         os.makedirs(os.path.join(self.tardir, element._get_project().name, element.normal_name), exist_ok=True)
 
@@ -56,6 +85,21 @@ class TarCache(ArtifactCache):
 
                 _Tar.archive(os.path.join(self.tardir, ref), key, temp)
 
+            self.cache_size = None
+            self.append_required_artifacts([element])
+
+    # update_atime():
+    #
+    # Update the access time of an element.
+    #
+    # Args:
+    #     element (Element): The Element to mark
+    #     key (str): The cache key to use
+    #
+    def update_atime(self, element, key):
+        path = _tarpath(element, key)
+        os.utime(os.path.join(self.tardir, path))
+
     def extract(self, element, key):
 
         fullname = self.get_artifact_fullname(element, key)
@@ -90,6 +134,21 @@ class TarCache(ArtifactCache):
 
         return dest
 
+    # get_cache_size()
+    #
+    # Return the artifact cache size.
+    #
+    # Returns:
+    #
+    # (int) The size of the artifact cache.
+    #
+    def calculate_cache_size(self):
+        if self.cache_size is None:
+            self.cache_size = utils._get_dir_size(self.tardir)
+            self.estimated_size = self.cache_size
+
+        return self.cache_size
+
 
 # _tarpath()
 #
diff --git a/buildstream/_context.py b/buildstream/_context.py
index 1a59af2b9..5cc7f434c 100644
--- a/buildstream/_context.py
+++ b/buildstream/_context.py
@@ -21,6 +21,7 @@ import os
 import datetime
 from collections import deque, Mapping
 from contextlib import contextmanager
+from . import utils
 from . import _cachekey
 from . import _signals
 from . import _site
@@ -62,6 +63,12 @@ class Context():
         # The locations from which to push and pull prebuilt artifacts
         self.artifact_cache_specs = []
 
+        # The artifact cache quota
+        self.cache_quota = None
+
+        # The lower threshold to which we aim to reduce the cache size
+        self.cache_lower_threshold = None
+
         # The directory to store build logs
         self.logdir = None
 
@@ -153,6 +160,7 @@ class Context():
         _yaml.node_validate(defaults, [
             'sourcedir', 'builddir', 'artifactdir', 'logdir',
             'scheduler', 'artifacts', 'logging', 'projects',
+            'cache'
         ])
 
         for directory in ['sourcedir', 'builddir', 'artifactdir', 'logdir']:
@@ -165,6 +173,53 @@ class Context():
             path = os.path.normpath(path)
             setattr(self, directory, path)
 
+        # Load quota configuration
+        # We need to find the first existing directory in the path of
+        # our artifactdir - the artifactdir may not have been created
+        # yet.
+        cache = _yaml.node_get(defaults, Mapping, 'cache')
+        _yaml.node_validate(cache, ['quota'])
+
+        artifactdir_volume = self.artifactdir
+        while not os.path.exists(artifactdir_volume):
+            artifactdir_volume = os.path.dirname(artifactdir_volume)
+
+        # We read and parse the cache quota as specified by the user
+        cache_quota = _yaml.node_get(cache, str, 'quota', default_value='infinity')
+        try:
+            cache_quota = utils._parse_size(cache_quota, artifactdir_volume)
+        except utils.UtilError as e:
+            raise LoadError(LoadErrorReason.INVALID_DATA,
+                            "{}\nPlease specify the value in bytes or as a % of full disk space.\n"
+                            "\nValid values are, for example: 800M 10G 1T 50%\n"
+                            .format(str(e))) from e
+
+        # If we are asked not to set a quota, we set it to the maximum
+        # disk space available minus a headroom of 2GB, such that we
+        # at least try to avoid raising Exceptions.
+        #
+        # Of course, we might still end up running out during a build
+        # if we end up writing more than 2G, but hey, this stuff is
+        # already really fuzzy.
+        #
+        if cache_quota is None:
+            stat = os.statvfs(artifactdir_volume)
+            # Again, the artifact directory may not yet have been
+            # created
+            if not os.path.exists(self.artifactdir):
+                cache_size = 0
+            else:
+                cache_size = utils._get_dir_size(self.artifactdir)
+            cache_quota = cache_size + stat.f_bsize * stat.f_bavail
+
+        if 'BST_TEST_SUITE' in os.environ:
+            headroom = 0
+        else:
+            headroom = 2e9
+
+        self.cache_quota = cache_quota - headroom
+        self.cache_lower_threshold = self.cache_quota / 2
+
         # Load artifact share configuration
         self.artifact_cache_specs = ArtifactCache.specs_from_config_node(defaults)
 
diff --git a/buildstream/_ostree.py b/buildstream/_ostree.py
index 238c6b4db..d0609605a 100644
--- a/buildstream/_ostree.py
+++ b/buildstream/_ostree.py
@@ -565,6 +565,8 @@ def list_artifacts(repo):
     ref_heads = os.path.join(repo.get_path().get_path(), 'refs', 'heads')
 
     # obtain list of <project>/<element>/<key>
+    # FIXME: ostree 2017.11+ supports a flag that would allow
+    #        listing only local refs
     refs = _list_all_refs(repo).keys()
 
     mtimes = []
diff --git a/buildstream/_pipeline.py b/buildstream/_pipeline.py
index 9f4504d3f..7f159c7ec 100644
--- a/buildstream/_pipeline.py
+++ b/buildstream/_pipeline.py
@@ -159,6 +159,8 @@ class Pipeline():
                 # Determine initial element state.
                 element._update_state()
 
+            self._artifacts.append_required_artifacts((e for e in self.dependencies(targets, Scope.ALL)))
+
     # dependencies()
     #
     # Generator function to iterate over elements and optionally
diff --git a/buildstream/_scheduler/jobs/cachesizejob.py b/buildstream/_scheduler/jobs/cachesizejob.py
new file mode 100644
index 000000000..897e896ca
--- /dev/null
+++ b/buildstream/_scheduler/jobs/cachesizejob.py
@@ -0,0 +1,91 @@
+#  Copyright (C) 2018 Codethink Limited
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Author:
+#        Tristan Daniël Maat <tristan.maat@codethink.co.uk>
+#
+import os
+from contextlib import contextmanager
+
+from .job import Job
+from ..._platform import Platform
+from ..._message import Message, MessageType
+
+
+class CacheSizeJob(Job):
+    def __init__(self, *args, complete_cb, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._complete_cb = complete_cb
+        self._cache = Platform._instance.artifactcache
+
+    def _child_process(self):
+        return self._cache.calculate_cache_size()
+
+    def _parent_complete(self, success, result):
+        self._cache._set_cache_size(result)
+        if self._complete_cb:
+            self._complete_cb(result)
+
+    @contextmanager
+    def _child_logging_enabled(self, logfile):
+        self._logfile = logfile.format(pid=os.getpid())
+        yield self._logfile
+        self._logfile = None
+
+    # _message():
+    #
+    # Sends a message to the frontend
+    #
+    # Args:
+    #    message_type (MessageType): The type of message to send
+    #    message (str): The message
+    #    kwargs: Remaining Message() constructor arguments
+    #
+    def _message(self, message_type, message, **kwargs):
+        args = dict(kwargs)
+        args['scheduler'] = True
+        self._scheduler.context.message(Message(None, message_type, message, **args))
+
+    def _child_log(self, message):
+        with open(self._logfile, 'a+') as log:
+            INDENT = "    "
+            EMPTYTIME = "--:--:--"
+
+            template = "[{timecode: <8}] {type: <7} {name: <15}: {message}"
+            detail = ''
+            if message.detail is not None:
+                template += "\n\n{detail}"
+                detail = message.detail.rstrip('\n')
+                detail = INDENT + INDENT.join(detail.splitlines(True))
+
+            timecode = EMPTYTIME
+            if message.message_type in (MessageType.SUCCESS, MessageType.FAIL):
+                hours, remainder = divmod(int(message.elapsed.total_seconds()), 60**2)
+                minutes, seconds = divmod(remainder, 60)
+                timecode = "{0:02d}:{1:02d}:{2:02d}".format(hours, minutes, seconds)
+
+            message_text = template.format(timecode=timecode,
+                                           type=message.message_type.upper(),
+                                           name='cache_size',
+                                           message=message.message,
+                                           detail=detail)
+
+            log.write('{}\n'.format(message_text))
+            log.flush()
+
+        return message
+
+    def _child_process_data(self):
+        return {}
diff --git a/buildstream/_scheduler/jobs/cleanupjob.py b/buildstream/_scheduler/jobs/cleanupjob.py
new file mode 100644
index 000000000..2de36803f
--- /dev/null
+++ b/buildstream/_scheduler/jobs/cleanupjob.py
@@ -0,0 +1,72 @@
+#  Copyright (C) 2018 Codethink Limited
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Author:
+#        Tristan Daniël Maat <tristan.maat@codethink.co.uk>
+#
+import os
+from contextlib import contextmanager
+
+from .job import Job
+from ..._platform import Platform
+from ..._message import Message
+
+
+class CleanupJob(Job):
+    def __init__(self, *args, complete_cb, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._complete_cb = complete_cb
+        self._cache = Platform._instance.artifactcache
+
+    def _child_process(self):
+        return self._cache.clean()
+
+    def _parent_complete(self, success, result):
+        self._cache._set_cache_size(result)
+        if self._complete_cb:
+            self._complete_cb()
+
+    @contextmanager
+    def _child_logging_enabled(self, logfile):
+        self._logfile = logfile.format(pid=os.getpid())
+        yield self._logfile
+        self._logfile = None
+
+    # _message():
+    #
+    # Sends a message to the frontend
+    #
+    # Args:
+    #    message_type (MessageType): The type of message to send
+    #    message (str): The message
+    #    kwargs: Remaining Message() constructor arguments
+    #
+    def _message(self, message_type, message, **kwargs):
+        args = dict(kwargs)
+        args['scheduler'] = True
+        self._scheduler.context.message(Message(None, message_type, message, **args))
+
+    def _child_log(self, message):
+        message.action_name = self.action_name
+
+        with open(self._logfile, 'a+') as log:
+            message_text = self._format_frontend_message(message, '[cleanup]')
+            log.write('{}\n'.format(message_text))
+            log.flush()
+
+        return message
+
+    def _child_process_data(self):
+        return {}
diff --git a/buildstream/_scheduler/queues/buildqueue.py b/buildstream/_scheduler/queues/buildqueue.py
index 7f8ac9e8f..376ef5ae2 100644
--- a/buildstream/_scheduler/queues/buildqueue.py
+++ b/buildstream/_scheduler/queues/buildqueue.py
@@ -51,10 +51,27 @@ class BuildQueue(Queue):
 
         return QueueStatus.READY
 
+    def _check_cache_size(self, job, element):
+        if not job.child_data:
+            return
+
+        artifact_size = job.child_data.get('artifact_size', False)
+
+        if artifact_size:
+            cache = element._get_artifact_cache()
+            cache._add_artifact_size(artifact_size)
+
+            if cache.get_approximate_cache_size() > self._scheduler.context.cache_quota:
+                self._scheduler._check_cache_size_real()
+
     def done(self, job, element, result, success):
 
         if success:
             # Inform element in main process that assembly is done
             element._assemble_done()
 
+        # This has to be done after _assemble_done, such that the
+        # element may register its cache key as required
+        self._check_cache_size(job, element)
+
         return True
diff --git a/buildstream/_scheduler/queues/pullqueue.py b/buildstream/_scheduler/queues/pullqueue.py
index efaa59ef3..430afc410 100644
--- a/buildstream/_scheduler/queues/pullqueue.py
+++ b/buildstream/_scheduler/queues/pullqueue.py
@@ -59,6 +59,11 @@ class PullQueue(Queue):
 
         element._pull_done()
 
+        # Build jobs will check the "approximate" size first. Since we
+        # do not get an artifact size from pull jobs, we have to
+        # actually check the cache size.
+        self._scheduler._check_cache_size_real()
+
         # Element._pull() returns True if it downloaded an artifact,
         # here we want to appear skipped if we did not download.
         return result
diff --git a/buildstream/_scheduler/queues/queue.py b/buildstream/_scheduler/queues/queue.py
index 8ca3ac063..ac20d3711 100644
--- a/buildstream/_scheduler/queues/queue.py
+++ b/buildstream/_scheduler/queues/queue.py
@@ -300,6 +300,8 @@ class Queue():
         # Update values that need to be synchronized in the main task
         # before calling any queue implementation
         self._update_workspaces(element, job)
+        if job.child_data:
+            element._get_artifact_cache().cache_size = job.child_data.get('cache_size')
 
         # Give the result of the job to the Queue implementor,
         # and determine if it should be considered as processed
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index bc182db32..aeb32931a 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -27,7 +27,8 @@ import datetime
 from contextlib import contextmanager
 
 # Local imports
-from .resources import Resources
+from .resources import Resources, ResourceType
+from .jobs import CacheSizeJob, CleanupJob
 
 
 # A decent return code for Scheduler.run()
@@ -312,6 +313,27 @@ class Scheduler():
         self.schedule_jobs(ready)
         self._sched()
 
+    def _run_cleanup(self, cache_size):
+        if cache_size and cache_size < self.context.cache_quota:
+            return
+
+        logpath = os.path.join(self.context.logdir, 'cleanup.{pid}.log')
+        job = CleanupJob(self, 'cleanup', logpath,
+                         resources=[ResourceType.CACHE,
+                                    ResourceType.PROCESS],
+                         exclusive_resources=[ResourceType.CACHE],
+                         complete_cb=None)
+        self.schedule_jobs([job])
+
+    def _check_cache_size_real(self):
+        logpath = os.path.join(self.context.logdir, 'cache_size.{pid}.log')
+        job = CacheSizeJob(self, 'cache_size', logpath,
+                           resources=[ResourceType.CACHE,
+                                      ResourceType.PROCESS],
+                           exclusive_resources=[ResourceType.CACHE],
+                           complete_cb=self._run_cleanup)
+        self.schedule_jobs([job])
+
     # _suspend_jobs()
     #
     # Suspend all ongoing jobs.
diff --git a/buildstream/data/userconfig.yaml b/buildstream/data/userconfig.yaml
index 6bb54ff9e..6f9f190a1 100644
--- a/buildstream/data/userconfig.yaml
+++ b/buildstream/data/userconfig.yaml
@@ -23,6 +23,14 @@ artifactdir: ${XDG_CACHE_HOME}/buildstream/artifacts
 logdir: ${XDG_CACHE_HOME}/buildstream/logs
 
 #
+#    Cache
+#
+cache:
+  # Size of the artifact cache - BuildStream will attempt to keep the
+  # artifact cache within this size.
+  quota: infinity
+
+#
 #    Scheduler
 #
 scheduler:
diff --git a/buildstream/element.py b/buildstream/element.py
index fc21f80b6..518fb59f7 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -225,6 +225,7 @@ class Element(Plugin):
         self.__staged_sources_directory = None  # Location where Element.stage_sources() was called
         self.__tainted = None                   # Whether the artifact is tainted and should not be shared
         self.__required = False                 # Whether the artifact is required in the current session
+        self.__artifact_size = None             # The size of data committed to the artifact cache
 
         # hash tables of loaded artifact metadata, hashed by key
         self.__metadata_keys = {}                     # Strong and weak keys for this key
@@ -1397,6 +1398,16 @@ class Element(Plugin):
             workspace.clear_running_files()
             self._get_context().get_workspaces().save_config()
 
+            # We also need to update the required artifacts, since
+            # workspaced dependencies do not have a fixed cache key
+            # when the build starts.
+            #
+            # This does *not* cause a race condition, because
+            # _assemble_done is called before a cleanup job may be
+            # launched.
+            #
+            self.__artifacts.append_required_artifacts([self])
+
     # _assemble():
     #
     # Internal method for running the entire build phase.
@@ -1524,6 +1535,7 @@ class Element(Plugin):
                 }), os.path.join(metadir, 'workspaced-dependencies.yaml'))
 
                 with self.timed_activity("Caching artifact"):
+                    self.__artifact_size = utils._get_dir_size(assembledir)
                     self.__artifacts.commit(self, assembledir, self.__get_cache_keys_for_commit())
 
             # Finally cleanup the build dir
@@ -1763,6 +1775,25 @@ class Element(Plugin):
         workspaces = self._get_context().get_workspaces()
         return workspaces.get_workspace(self._get_full_name())
 
+    # _get_artifact_size()
+    #
+    # Get the size of the artifact produced by this element in the
+    # current pipeline - if this element has not been assembled or
+    # pulled, this will be None.
+    #
+    # Note that this is the size of an artifact *before* committing it
+    # to the cache, the size on disk may differ. It can act as an
+    # approximate guide for when to do a proper size calculation.
+    #
+    # Returns:
+    #    (int|None): The size of the artifact
+    #
+    def _get_artifact_size(self):
+        return self.__artifact_size
+
+    def _get_artifact_cache(self):
+        return self.__artifacts
+
     # _write_script():
     #
     # Writes a script to the given directory.
diff --git a/buildstream/utils.py b/buildstream/utils.py
index b81a6c852..e8270d82f 100644
--- a/buildstream/utils.py
+++ b/buildstream/utils.py
@@ -96,7 +96,7 @@ class FileListResult():
         return ret
 
 
-def list_relative_paths(directory):
+def list_relative_paths(directory, *, list_dirs=True):
     """A generator for walking directory relative paths
 
     This generator is useful for checking the full manifest of
@@ -110,6 +110,7 @@ def list_relative_paths(directory):
 
     Args:
        directory (str): The directory to list files in
+       list_dirs (bool): Whether to list directories
 
     Yields:
        Relative filenames in `directory`
@@ -136,15 +137,16 @@ def list_relative_paths(directory):
         # subdirectories in the walked `dirpath`, so we extract
         # these symlinks from `dirnames`
         #
-        for d in dirnames:
-            fullpath = os.path.join(dirpath, d)
-            if os.path.islink(fullpath):
-                yield os.path.join(basepath, d)
+        if list_dirs:
+            for d in dirnames:
+                fullpath = os.path.join(dirpath, d)
+                if os.path.islink(fullpath):
+                    yield os.path.join(basepath, d)
 
         # We've decended into an empty directory, in this case we
         # want to include the directory itself, but not in any other
         # case.
-        if not filenames:
+        if list_dirs and not filenames:
             yield relpath
 
         # List the filenames in the walked directory
@@ -536,6 +538,76 @@ def save_file_atomic(filename, mode='w', *, buffering=-1, encoding=None,
         raise
 
 
+# _get_dir_size():
+#
+# Get the disk usage of a given directory in bytes.
+#
+# Arguments:
+#     (str) The path whose size to check.
+#
+# Returns:
+#     (int) The size on disk in bytes.
+#
+def _get_dir_size(path):
+    path = os.path.abspath(path)
+
+    def get_size(path):
+        total = 0
+
+        for f in os.scandir(path):
+            total += f.stat(follow_symlinks=False).st_size
+
+            if f.is_dir(follow_symlinks=False):
+                total += get_size(f.path)
+
+        return total
+
+    return get_size(path)
+
+
+# _parse_size():
+#
+# Convert a string representing data size to a number of
+# bytes. E.g. "2K" -> 2048.
+#
+# This uses the same format as systemd's
+# [resource-control](https://www.freedesktop.org/software/systemd/man/systemd.resource-control.html#).
+#
+# Arguments:
+#     size (str) The string to parse
+#     volume (str) A path on the volume to consider for percentage
+#                  specifications
+#
+# Returns:
+#     (int|None) The number of bytes, or None if 'infinity' was specified.
+#
+# Raises:
+#     UtilError if the string is not a valid data size.
+#
+def _parse_size(size, volume):
+    if size == 'infinity':
+        return None
+
+    matches = re.fullmatch(r'([0-9]+\.?[0-9]*)([KMGT%]?)', size)
+    if matches is None:
+        raise UtilError("{} is not a valid data size.".format(size))
+
+    num, unit = matches.groups()
+
+    if unit == '%':
+        num = float(num)
+        if num > 100:
+            raise UtilError("{}% is not a valid percentage value.".format(num))
+
+        stat_ = os.statvfs(volume)
+        disk_size = stat_.f_blocks * stat_.f_bsize
+
+        return disk_size * (num / 100)
+
+    units = ('', 'K', 'M', 'G', 'T')
+    return int(num) * 1024**units.index(unit)
+
+
 # A sentinel to be used as a default argument for functions that need
 # to distinguish between a kwarg set to None and an unset kwarg.
 _sentinel = object()
diff --git a/tests/artifactcache/expiry.py b/tests/artifactcache/expiry.py
new file mode 100644
index 000000000..4c741054b
--- /dev/null
+++ b/tests/artifactcache/expiry.py
@@ -0,0 +1,264 @@
+import os
+
+import pytest
+
+from buildstream import _yaml
+from buildstream._exceptions import ErrorDomain, LoadErrorReason
+
+from tests.testutils import cli
+
+
+DATA_DIR = os.path.join(
+    os.path.dirname(os.path.realpath(__file__)),
+    "expiry"
+)
+
+
+def create_element(name, path, dependencies, size):
+    os.makedirs(path, exist_ok=True)
+
+    # Create a file to be included in this element's artifact
+    with open(os.path.join(path, name + '_data'), 'wb+') as f:
+        f.write(os.urandom(size))
+
+    element = {
+        'kind': 'import',
+        'sources': [
+            {
+                'kind': 'local',
+                'path': os.path.join(path, name + '_data')
+            }
+        ],
+        'depends': dependencies
+    }
+    _yaml.dump(element, os.path.join(path, name))
+
+
+# Ensure that the cache successfully removes an old artifact if we do
+# not have enough space left.
+@pytest.mark.datafiles(DATA_DIR)
+def test_artifact_expires(cli, datafiles, tmpdir):
+    project = os.path.join(datafiles.dirname, datafiles.basename)
+    element_path = os.path.join(project, 'elements')
+    cache_location = os.path.join(project, 'cache', 'artifacts', 'ostree')
+    checkout = os.path.join(project, 'checkout')
+
+    cli.configure({
+        'cache': {
+            'quota': 10000000,
+        }
+    })
+
+    # Create an element that uses almost the entire cache (an empty
+    # ostree cache starts at about ~10KiB, so we need a bit of a
+    # buffer)
+    create_element('target.bst', element_path, [], 6000000)
+    res = cli.run(project=project, args=['build', 'target.bst'])
+    res.assert_success()
+
+    assert cli.get_element_state(project, 'target.bst') == 'cached'
+
+    # Our cache should now be almost full. Let's create another
+    # artifact and see if we can cause buildstream to delete the old
+    # one.
+    create_element('target2.bst', element_path, [], 6000000)
+    res = cli.run(project=project, args=['build', 'target2.bst'])
+    res.assert_success()
+
+    # Check that the correct element remains in the cache
+    assert cli.get_element_state(project, 'target.bst') != 'cached'
+    assert cli.get_element_state(project, 'target2.bst') == 'cached'
+
+
+# Ensure that we don't end up deleting the whole cache (or worse) if
+# we try to store an artifact that is too large to fit in the quota.
+@pytest.mark.parametrize('size', [
+    # Test an artifact that is obviously too large
+    (500000),
+    # Test an artifact that might be too large due to slight overhead
+    # of storing stuff in ostree
+    (399999)
+])
+@pytest.mark.datafiles(DATA_DIR)
+def test_artifact_too_large(cli, datafiles, tmpdir, size):
+    project = os.path.join(datafiles.dirname, datafiles.basename)
+    element_path = os.path.join(project, 'elements')
+
+    cli.configure({
+        'cache': {
+            'quota': 400000
+        }
+    })
+
+    # Create an element whose artifact is too large
+    create_element('target.bst', element_path, [], size)
+    res = cli.run(project=project, args=['build', 'target.bst'])
+    res.assert_main_error(ErrorDomain.STREAM, None)
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_expiry_order(cli, datafiles, tmpdir):
+    project = os.path.join(datafiles.dirname, datafiles.basename)
+    element_path = os.path.join(project, 'elements')
+    cache_location = os.path.join(project, 'cache', 'artifacts', 'ostree')
+    checkout = os.path.join(project, 'workspace')
+
+    cli.configure({
+        'cache': {
+            'quota': 9000000
+        }
+    })
+
+    # Create an artifact
+    create_element('dep.bst', element_path, [], 2000000)
+    res = cli.run(project=project, args=['build', 'dep.bst'])
+    res.assert_success()
+
+    # Create another artifact
+    create_element('unrelated.bst', element_path, [], 2000000)
+    res = cli.run(project=project, args=['build', 'unrelated.bst'])
+    res.assert_success()
+
+    # And build something else
+    create_element('target.bst', element_path, [], 2000000)
+    res = cli.run(project=project, args=['build', 'target.bst'])
+    res.assert_success()
+
+    create_element('target2.bst', element_path, [], 2000000)
+    res = cli.run(project=project, args=['build', 'target2.bst'])
+    res.assert_success()
+
+    # Now extract dep.bst
+    res = cli.run(project=project, args=['checkout', 'dep.bst', checkout])
+    res.assert_success()
+
+    # Finally, build something that will cause the cache to overflow
+    create_element('expire.bst', element_path, [], 2000000)
+    res = cli.run(project=project, args=['build', 'expire.bst'])
+    res.assert_success()
+
+    # While dep.bst was the first element to be created, it should not
+    # have been removed.
+    # Note that buildstream will reduce the cache to 50% of the
+    # original size - we therefore remove multiple elements.
+
+    assert (tuple(cli.get_element_state(project, element) for element in
+                  ('unrelated.bst', 'target.bst', 'target2.bst', 'dep.bst', 'expire.bst')) ==
+            ('buildable', 'buildable', 'buildable', 'cached', 'cached', ))
+
+
+# Ensure that we don't accidentally remove an artifact from something
+# in the current build pipeline, because that would be embarassing,
+# wouldn't it?
+@pytest.mark.datafiles(DATA_DIR)
+def test_keep_dependencies(cli, datafiles, tmpdir):
+    project = os.path.join(datafiles.dirname, datafiles.basename)
+    element_path = os.path.join(project, 'elements')
+    cache_location = os.path.join(project, 'cache', 'artifacts', 'ostree')
+
+    cli.configure({
+        'cache': {
+            'quota': 10000000
+        }
+    })
+
+    # Create a pretty big dependency
+    create_element('dependency.bst', element_path, [], 5000000)
+    res = cli.run(project=project, args=['build', 'dependency.bst'])
+    res.assert_success()
+
+    # Now create some other unrelated artifact
+    create_element('unrelated.bst', element_path, [], 4000000)
+    res = cli.run(project=project, args=['build', 'unrelated.bst'])
+    res.assert_success()
+
+    # Check that the correct element remains in the cache
+    assert cli.get_element_state(project, 'dependency.bst') == 'cached'
+    assert cli.get_element_state(project, 'unrelated.bst') == 'cached'
+
+    # We try to build an element which depends on the LRU artifact,
+    # and could therefore fail if we didn't make sure dependencies
+    # aren't removed.
+    #
+    # Since some artifact caches may implement weak cache keys by
+    # duplicating artifacts (bad!) we need to make this equal in size
+    # or smaller than half the size of its dependencies.
+    #
+    create_element('target.bst', element_path, ['dependency.bst'], 2000000)
+    res = cli.run(project=project, args=['build', 'target.bst'])
+    res.assert_success()
+
+    assert cli.get_element_state(project, 'unrelated.bst') != 'cached'
+    assert cli.get_element_state(project, 'dependency.bst') == 'cached'
+    assert cli.get_element_state(project, 'target.bst') == 'cached'
+
+
+# Assert that we never delete a dependency required for a build tree
+@pytest.mark.datafiles(DATA_DIR)
+def test_never_delete_dependencies(cli, datafiles, tmpdir):
+    project = os.path.join(datafiles.dirname, datafiles.basename)
+    element_path = os.path.join(project, 'elements')
+
+    cli.configure({
+        'cache': {
+            'quota': 10000000
+        }
+    })
+
+    # Create a build tree
+    create_element('dependency.bst', element_path, [], 8000000)
+    create_element('related.bst', element_path, ['dependency.bst'], 8000000)
+    create_element('target.bst', element_path, ['related.bst'], 8000000)
+    create_element('target2.bst', element_path, ['target.bst'], 8000000)
+
+    # We try to build this pipeline, but it's too big for the
+    # cache. Since all elements are required, the build should fail.
+    res = cli.run(project=project, args=['build', 'target2.bst'])
+    res.assert_main_error(ErrorDomain.STREAM, None)
+
+    assert cli.get_element_state(project, 'dependency.bst') == 'cached'
+
+    # This is *technically* above the cache limit. BuildStream accepts
+    # some fuzziness, since it's hard to assert that we don't create
+    # an artifact larger than the cache quota. We would have to remove
+    # the artifact after-the-fact, but since it is required for the
+    # current build and nothing broke yet, it's nicer to keep it
+    # around.
+    #
+    # This scenario is quite unlikely, and the cache overflow will be
+    # resolved if the user does something about it anyway.
+    #
+    assert cli.get_element_state(project, 'related.bst') == 'cached'
+
+    assert cli.get_element_state(project, 'target.bst') != 'cached'
+    assert cli.get_element_state(project, 'target2.bst') != 'cached'
+
+
+# Ensure that only valid cache quotas make it through the loading
+# process.
+@pytest.mark.parametrize("quota,success", [
+    ("1", True),
+    ("1K", True),
+    ("50%", True),
+    ("infinity", True),
+    ("0", True),
+    ("-1", False),
+    ("pony", False),
+    ("200%", False)
+])
+@pytest.mark.datafiles(DATA_DIR)
+def test_invalid_cache_quota(cli, datafiles, tmpdir, quota, success):
+    project = os.path.join(datafiles.dirname, datafiles.basename)
+    element_path = os.path.join(project, 'elements')
+
+    cli.configure({
+        'cache': {
+            'quota': quota,
+        }
+    })
+
+    res = cli.run(project=project, args=['workspace', 'list'])
+    if success:
+        res.assert_success()
+    else:
+        res.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.INVALID_DATA)
diff --git a/tests/artifactcache/expiry/project.conf b/tests/artifactcache/expiry/project.conf
new file mode 100644
index 000000000..18db7dab7
--- /dev/null
+++ b/tests/artifactcache/expiry/project.conf
@@ -0,0 +1,14 @@
+# Project config for cache expiry test
+name: test
+element-path: elements
+aliases:
+  project_dir: file://{project_dir}
+options:
+  linux:
+    type: bool
+    description: Whether to expect a linux platform
+    default: True
+split-rules:
+  test:
+    - |
+      /tests/*
diff --git a/tests/frontend/push.py b/tests/frontend/push.py
index ca46b0447..cdd6ff1a2 100644
--- a/tests/frontend/push.py
+++ b/tests/frontend/push.py
@@ -2,7 +2,7 @@ import os
 import shutil
 import pytest
 from collections import namedtuple
-from unittest.mock import MagicMock
+from unittest.mock import patch
 
 from buildstream._exceptions import ErrorDomain
 from tests.testutils import cli, create_artifact_share, create_element_size
@@ -17,6 +17,20 @@ DATA_DIR = os.path.join(
 )
 
 
+# The original result of os.statvfs so that we can mock it
+NORMAL_STAT = os.statvfs('/')
+
+
+def stat_tuple():
+    stat = NORMAL_STAT
+    bsize = stat.f_bsize
+
+    fields = [var for var in dir(stat) if isinstance(getattr(stat, var), int)][0:stat.n_fields]
+    statvfs_result = namedtuple('statvfs_result', ' '.join(fields))
+
+    return statvfs_result(*[getattr(stat, var) for var in fields])
+
+
 # Assert that a given artifact is in the share
 #
 def assert_shared(cli, share, project, element_name):
@@ -205,6 +219,7 @@ def test_push_after_pull(cli, tmpdir, datafiles):
 # Ensure that when an artifact's size exceeds available disk space
 # the least recently pushed artifact is deleted in order to make room for
 # the incoming artifact.
+@pytest.mark.xfail
 @pytest.mark.datafiles(DATA_DIR)
 def test_artifact_expires(cli, datafiles, tmpdir):
     project = os.path.join(datafiles.dirname, datafiles.basename)
@@ -213,13 +228,6 @@ def test_artifact_expires(cli, datafiles, tmpdir):
     # Create an artifact share (remote artifact cache) in the tmpdir/artifactshare
     share = create_artifact_share(os.path.join(str(tmpdir), 'artifactshare'))
 
-    # Mock the os.statvfs() call to return a named tuple which emulates an
-    # os.statvfs_result object
-    statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize')
-    os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(10e9),
-                                                       f_bfree=(int(12e6) + int(2e9)),
-                                                       f_bsize=1))
-
     # Configure bst to push to the cache
     cli.configure({
         'artifacts': {'url': share.repo, 'push': True},
@@ -227,16 +235,26 @@ def test_artifact_expires(cli, datafiles, tmpdir):
 
     # Create and build an element of 5 MB
     create_element_size('element1.bst', element_path, [], int(5e6))  # [] => no deps
-    result = cli.run(project=project, args=['build', 'element1.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'element1.bst'])
     result.assert_success()
 
     # Create and build an element of 5 MB
     create_element_size('element2.bst', element_path, [], int(5e6))  # [] => no deps
-    result = cli.run(project=project, args=['build', 'element2.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'element2.bst'])
     result.assert_success()
 
+    # Mock the os.statvfs() call to return a named tuple which emulates an
+    # os.statvfs_result object
+    free_space = int(12e6)
+
+    free = stat_tuple()._replace(f_blocks=int(10e9), f_bfree=free_space + int(2e9), f_bsize=1)
+    with patch('os.statvfs', return_value=free):
+        result = cli.run(project=project, args=['push', 'element1.bst', 'element2.bst'])
+        result.assert_success()
+
     # update the share
     share.update_summary()
+    free_space -= 10e6
 
     # check that element's 1 and 2 are cached both locally and remotely
     assert cli.get_element_state(project, 'element1.bst') == 'cached'
@@ -244,18 +262,19 @@ def test_artifact_expires(cli, datafiles, tmpdir):
     assert cli.get_element_state(project, 'element2.bst') == 'cached'
     assert_shared(cli, share, project, 'element2.bst')
 
-    # update mocked available disk space now that two 5 MB artifacts have been added
-    os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(10e9),
-                                                       f_bfree=(int(2e6) + int(2e9)),
-                                                       f_bsize=1))
-
     # Create and build another element of 5 MB (This will exceed the free disk space available)
     create_element_size('element3.bst', element_path, [], int(5e6))
-    result = cli.run(project=project, args=['build', 'element3.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'element3.bst'])
     result.assert_success()
 
+    free = stat_tuple()._replace(f_blocks=int(10e9), f_bfree=free_space + int(2e9), f_bsize=1)
+    with patch('os.statvfs', return_value=free):
+        result = cli.run(project=project, args=['push', 'element3.bst'])
+        result.assert_success()
+
     # update the share
     share.update_summary()
+    free_space -= 5e6
 
     # Ensure it is cached both locally and remotely
     assert cli.get_element_state(project, 'element3.bst') == 'cached'
@@ -269,6 +288,7 @@ def test_artifact_expires(cli, datafiles, tmpdir):
 
 # Test that a large artifact, whose size exceeds the quota, is not pushed
 # to the remote share
+@pytest.mark.xfail
 @pytest.mark.datafiles(DATA_DIR)
 def test_artifact_too_large(cli, datafiles, tmpdir):
     project = os.path.join(datafiles.dirname, datafiles.basename)
@@ -277,12 +297,6 @@ def test_artifact_too_large(cli, datafiles, tmpdir):
     # Create an artifact share (remote cache) in tmpdir/artifactshare
     share = create_artifact_share(os.path.join(str(tmpdir), 'artifactshare'))
 
-    # Mock a file system with 5 MB total space
-    statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize')
-    os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(5e6) + int(2e9),
-                                                       f_bfree=(int(5e6) + int(2e9)),
-                                                       f_bsize=1))
-
     # Configure bst to push to the remote cache
     cli.configure({
         'artifacts': {'url': share.repo, 'push': True},
@@ -290,14 +304,20 @@ def test_artifact_too_large(cli, datafiles, tmpdir):
 
     # Create and push a 3MB element
     create_element_size('small_element.bst', element_path, [], int(3e6))
-    result = cli.run(project=project, args=['build', 'small_element.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'small_element.bst'])
     result.assert_success()
 
     # Create and try to push a 6MB element.
     create_element_size('large_element.bst', element_path, [], int(6e6))
-    result = cli.run(project=project, args=['build', 'large_element.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'large_element.bst'])
     result.assert_success()
 
+    # Mock a file system with 5 MB total space
+    free = stat_tuple()._replace(f_blocks=int(5e6) + int(2e9), f_bfree=int(5e6) + int(2e9), f_bsize=1)
+    with patch('os.statvfs', return_value=free):
+        result = cli.run(project=project, args=['push', 'small_element.bst', 'large_element.bst'])
+        result.assert_success()
+
     # update the cache
     share.update_summary()
 
@@ -323,12 +343,6 @@ def test_recently_pulled_artifact_does_not_expire(cli, datafiles, tmpdir):
     # Create an artifact share (remote cache) in tmpdir/artifactshare
     share = create_artifact_share(os.path.join(str(tmpdir), 'artifactshare'))
 
-    # Mock a file system with 12 MB free disk space
-    statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize')
-    os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(10e9) + int(2e9),
-                                                       f_bfree=(int(12e6) + int(2e9)),
-                                                       f_bsize=1))
-
     # Configure bst to push to the cache
     cli.configure({
         'artifacts': {'url': share.repo, 'push': True},
@@ -336,14 +350,23 @@ def test_recently_pulled_artifact_does_not_expire(cli, datafiles, tmpdir):
 
     # Create and build 2 elements, each of 5 MB.
     create_element_size('element1.bst', element_path, [], int(5e6))
-    result = cli.run(project=project, args=['build', 'element1.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'element1.bst'])
     result.assert_success()
 
     create_element_size('element2.bst', element_path, [], int(5e6))
-    result = cli.run(project=project, args=['build', 'element2.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'element2.bst'])
     result.assert_success()
 
+    # Mock a file system with 12 MB free disk space
+    free_space = int(12e6)
+
+    free = stat_tuple()._replace(f_blocks=int(10e9) + int(2e9), f_bfree=free_space + int(2e9), f_bsize=1)
+    with patch('os.statvfs', return_value=free):
+        result = cli.run(project=project, args=['push', 'element1.bst', 'element2.bst'])
+        result.assert_success()
+
     share.update_summary()
+    free_space -= int(10e6)
 
     # Ensure they are cached locally
     assert cli.get_element_state(project, 'element1.bst') == 'cached'
@@ -367,10 +390,16 @@ def test_recently_pulled_artifact_does_not_expire(cli, datafiles, tmpdir):
 
     # Create and build the element3 (of 5 MB)
     create_element_size('element3.bst', element_path, [], int(5e6))
-    result = cli.run(project=project, args=['build', 'element3.bst'])
+    result = cli.run(project=project, args=['--pushers', '0', 'build', 'element3.bst'])
     result.assert_success()
 
+    free = stat_tuple()._replace(f_blocks=int(10e9) + int(2e9), f_bfree=free_space + int(2e9), f_bsize=1)
+    with patch('os.statvfs', return_value=free):
+        result = cli.run(project=project, args=['push', 'element3.bst'])
+        result.assert_success()
+
     share.update_summary()
+    free_space -= 5e6
 
     # Make sure it's cached locally and remotely
     assert cli.get_element_state(project, 'element3.bst') == 'cached'