source.py: Remove the reliance on consistency to get whether a source is cached

This removes the need to use consistency in Sources, by asking
explicitely whether the source is cached or not.

This introduces a new public method on source: `is_cached` that needs
implementation and that should return whether the source has a local
copy or not.

- On fetch, also reset whether the source was cached or set if as
  cached when we know it was.

- Validate the cache's source after fetching it

  This doesn't need to be run in the scheduler's process and can be
  offloaded to the child, which will allow better multiprocessing

10 files changed, 93 insertions, 18 deletions

diff --git a/src/buildstream/_gitsourcebase.py b/src/buildstream/_gitsourcebase.py
index ce2ef9731..125d34df4 100644
--- a/src/buildstream/_gitsourcebase.py
+++ b/src/buildstream/_gitsourcebase.py
@@ -527,6 +527,9 @@ class _GitSourceBase(Source):
     def is_resolved(self):
         return self.mirror.ref is not None
 
+    def is_cached(self):
+        return self._have_all_refs()
+
     def load_ref(self, node):
         self.mirror.ref = node.get_str("ref", None)
         self.mirror.tags = self._load_tags(node)
diff --git a/src/buildstream/_scheduler/queues/fetchqueue.py b/src/buildstream/_scheduler/queues/fetchqueue.py
index 8123b7611..18bf392d3 100644
--- a/src/buildstream/_scheduler/queues/fetchqueue.py
+++ b/src/buildstream/_scheduler/queues/fetchqueue.py
@@ -66,13 +66,7 @@ class FetchQueue(Queue):
         if status is JobStatus.FAIL:
             return
 
-        element._fetch_done()
-
-        # Successful fetch, we must be CACHED or in the sourcecache
-        if self._should_fetch_original:
-            assert element._has_all_sources_cached()
-        else:
-            assert element._has_all_sources_in_source_cache()
+        element._fetch_done(self._should_fetch_original)
 
     def register_pending_element(self, element):
         # Set a "can_query_cache" callback for an element not yet ready
diff --git a/src/buildstream/element.py b/src/buildstream/element.py
index f28fdf789..a5c879ba4 100644
--- a/src/buildstream/element.py
+++ b/src/buildstream/element.py
@@ -1737,15 +1737,12 @@ class Element(Plugin):
     #
     # Indicates that fetching the sources for this element has been done.
     #
-    def _fetch_done(self):
-        # We are not updating the state recursively here since fetching can
-        # never end up in updating them.
-
-        # Fetching changes the source state from RESOLVED to CACHED
-        # Fetching cannot change the source state from INCONSISTENT to CACHED because
-        # we prevent fetching when it's INCONSISTENT.
-        # Therefore, only the source state will change.
-        self.__update_source_state()
+    # Args:
+    #   fetched_original (bool): Whether the original sources had been asked (and fetched) or not
+    #
+    def _fetch_done(self, fetched_original):
+        for source in self.__sources:
+            source._fetch_done(fetched_original)
 
     # _pull_pending()
     #
diff --git a/src/buildstream/plugins/sources/_downloadablefilesource.py b/src/buildstream/plugins/sources/_downloadablefilesource.py
index 4e43ee3e3..581d32e7d 100644
--- a/src/buildstream/plugins/sources/_downloadablefilesource.py
+++ b/src/buildstream/plugins/sources/_downloadablefilesource.py
@@ -88,6 +88,9 @@ class DownloadableFileSource(Source):
     def get_unique_key(self):
         return [self.original_url, self.ref]
 
+    def is_cached(self) -> bool:
+        return os.path.isfile(self._get_mirror_file())
+
     def get_consistency(self):
         if self.ref is None:
             return Consistency.INCONSISTENT
diff --git a/src/buildstream/plugins/sources/bzr.py b/src/buildstream/plugins/sources/bzr.py
index 30ce55585..f9d95fca2 100644
--- a/src/buildstream/plugins/sources/bzr.py
+++ b/src/buildstream/plugins/sources/bzr.py
@@ -81,6 +81,10 @@ class BzrSource(Source):
     def get_unique_key(self):
         return [self.original_url, self.tracking, self.ref]
 
+    def is_cached(self):
+        with self._locked():
+            return self._check_ref()
+
     def get_consistency(self):
         if self.ref is None or self.tracking is None:
             return Consistency.INCONSISTENT
diff --git a/src/buildstream/plugins/sources/local.py b/src/buildstream/plugins/sources/local.py
index f4a7270e6..206dcb8c1 100644
--- a/src/buildstream/plugins/sources/local.py
+++ b/src/buildstream/plugins/sources/local.py
@@ -67,6 +67,9 @@ class LocalSource(Source):
     def is_resolved(self):
         return True
 
+    def is_cached(self):
+        return True
+
     # We dont have a ref, we're a local file...
     def load_ref(self, node):
         pass
diff --git a/src/buildstream/plugins/sources/patch.py b/src/buildstream/plugins/sources/patch.py
index cc3e6e2c4..cf6ce99cc 100644
--- a/src/buildstream/plugins/sources/patch.py
+++ b/src/buildstream/plugins/sources/patch.py
@@ -70,6 +70,9 @@ class PatchSource(Source):
     def is_resolved(self):
         return True
 
+    def is_cached(self):
+        return True
+
     def get_consistency(self):
         return Consistency.CACHED
 
diff --git a/src/buildstream/plugins/sources/pip.py b/src/buildstream/plugins/sources/pip.py
index 2c9773787..d45ef70c7 100644
--- a/src/buildstream/plugins/sources/pip.py
+++ b/src/buildstream/plugins/sources/pip.py
@@ -136,6 +136,9 @@ class PipSource(Source):
     def get_unique_key(self):
         return [self.original_url, self.ref]
 
+    def is_cached(self):
+        return os.path.exists(self._mirror) and os.listdir(self._mirror)
+
     def get_consistency(self):
         if not self.ref:
             return Consistency.INCONSISTENT
diff --git a/src/buildstream/plugins/sources/workspace.py b/src/buildstream/plugins/sources/workspace.py
index 88c32f754..5cea85dc2 100644
--- a/src/buildstream/plugins/sources/workspace.py
+++ b/src/buildstream/plugins/sources/workspace.py
@@ -69,6 +69,9 @@ class WorkspaceSource(Source):
     def preflight(self) -> None:
         pass  # pragma: nocover
 
+    def is_cached(self):
+        return True
+
     def is_resolved(self):
         return os.path.exists(self._get_local_path())
 
diff --git a/src/buildstream/source.py b/src/buildstream/source.py
index 46922b685..b7934374c 100644
--- a/src/buildstream/source.py
+++ b/src/buildstream/source.py
@@ -379,6 +379,8 @@ class Source(Plugin):
         self._configure(self.__config)
         self.__digest = None
 
+        self.__is_cached = None
+
     COMMON_CONFIG_KEYS = ["kind", "directory"]
     """Common source config keys
 
@@ -569,6 +571,19 @@ class Source(Plugin):
         *Since: 1.4*
         """
 
+    def is_cached(self) -> bool:
+        """Get whether the source has a local copy of its data.
+
+        This method is guaranteed to only be called whenever
+        :func:`Source.is_resolved() <buildstream.source.Source.is_resolved>`
+        returns `True`.
+
+        Returns: whether the source is cached locally or not.
+
+        *Since: 1.91.4*
+        """
+        raise ImplError("Source plugin '{}' does not implement is_cached()".format(self.get_kind()))
+
     #############################################################
     #                       Public Methods                      #
     #############################################################
@@ -802,7 +817,36 @@ class Source(Plugin):
     # Get whether the source is cached by the source plugin
     #
     def _is_cached(self):
-        return self.__consistency >= Consistency.CACHED
+        if self.__is_cached is None:
+            # We guarantee we only ever call this when we are resolved.
+            assert self.is_resolved()
+
+            # Set to 'False' on the first call, this prevents throwing multiple errors if the
+            # plugin throws exception when we display the end result pipeline.
+            # Otherwise, the summary would throw a second exception and we would not
+            # have a nice error reporting.
+            self.__is_cached = False
+
+            try:
+                self.__is_cached = self.is_cached()  # pylint: disable=assignment-from-no-return
+            except SourceError:
+                # SourceErrors should be preserved so that the
+                # plugin can communicate real error cases.
+                raise
+            except Exception as err:  # pylint: broad-except
+                # Generic errors point to bugs in the plugin, so
+                # we need to catch them and make sure they do not
+                # cause stacktraces
+
+                raise PluginError(
+                    "Source plugin '{}' failed to check its cached state: {}".format(self.get_kind(), err),
+                    reason="source-bug",
+                )
+
+            if self.__is_cached:
+                self.validate_cache()
+
+        return self.__is_cached
 
     # Wrapper function around plugin provided fetch method
     #
@@ -819,6 +863,24 @@ class Source(Plugin):
         else:
             self.__do_fetch()
 
+        self.validate_cache()
+
+    # _fetch_done()
+    #
+    # Indicates that fetching the source has been done.
+    #
+    # Args:
+    #   fetched_original (bool): Whether the original sources had been asked (and fetched) or not
+    #
+    def _fetch_done(self, fetched_original):
+        if fetched_original:
+            # The original was fetched, we know we are cached
+            self.__is_cached = True
+        else:
+            # The original was not requested, we might or might not be cached
+            # Don't recompute, but allow recomputation later if needed
+            self.__is_cached = None
+
     # Wrapper for stage() api which gives the source
     # plugin a fully constructed path considering the
     # 'directory' option
@@ -1429,7 +1491,7 @@ class Source(Plugin):
             # previous sources should never be in an inconsistent state
             assert src.is_resolved()
 
-            if src.get_consistency() == Consistency.RESOLVED:
+            if not src._is_cached():
                 src._fetch(previous_sources[0:index])