Move handling of the source `directory` configuration to ElementSourcesjuerg/element-source-cache

The `directory` value determines where a source is staged within the
build root of an element, however, it does not directly affect
individual sources.

With this change the sources will individually be cached in CAS
independent of the value of `directory`. `ElementSources` will use the
value of `directory` when staging all element sources into the build
root.

This results in a cache key change as the `directory` value is moved
from the unique key of individual sources to the unique key of
`ElementSources`.

This is in preparation for #1274.

2 files changed, 92 insertions, 100 deletions

diff --git a/src/buildstream/_elementsources.py b/src/buildstream/_elementsources.py
index 5fc412f6b..c030591f8 100644
--- a/src/buildstream/_elementsources.py
+++ b/src/buildstream/_elementsources.py
@@ -15,6 +15,8 @@
 #  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/>.
 
+import os
+from contextlib import contextmanager
 from typing import TYPE_CHECKING, Iterator
 
 from . import _cachekey, utils
@@ -48,10 +50,6 @@ class ElementSources:
         self._cache_key = None  # Our cached cache key
         self._proto = None  # The cached Source proto
 
-        # the index of the last source in this element that requires previous
-        # sources for staging
-        self._last_source_requires_previous_idx = None
-
     # get_project():
     #
     # Return the project associated with this object
@@ -92,9 +90,15 @@ class ElementSources:
     #
     def track(self, workspace):
         refs = []
-        for index, source in enumerate(self._sources):
+        for source in self._sources:
             old_ref = source.get_ref()
-            new_ref = source._track(self._sources[0:index])
+
+            if source.BST_REQUIRES_PREVIOUS_SOURCES_TRACK:
+                with self._stage_previous_sources(source) as staging_directory:
+                    new_ref = source._track(previous_sources_dir=staging_directory)
+            else:
+                new_ref = source._track()
+
             refs.append((source._unique_id, new_ref))
 
             # Complimentary warning that the new ref will be unused.
@@ -183,7 +187,14 @@ class ElementSources:
     #
     def init_workspace(self, directory: str):
         for source in self.sources():
-            source._init_workspace(directory)
+            if source._directory:
+                srcdir = os.path.join(directory, source._directory)
+            else:
+                srcdir = directory
+
+            os.makedirs(srcdir, exist_ok=True)
+
+            source._init_workspace(srcdir)
 
     # fetch():
     #
@@ -210,13 +221,16 @@ class ElementSources:
     #
     # Args:
     #   fetch_original (bool): Always fetch original source
+    #   stop (Source): Only fetch sources listed before this source
     #
     # Raises:
     #    SourceError: If one of the element sources has an error
     #
-    def fetch_sources(self, *, fetch_original=False):
-        previous_sources = []
+    def fetch_sources(self, *, fetch_original=False, stop=None):
         for source in self._sources:
+            if source == stop:
+                break
+
             if (
                 fetch_original
                 or source.BST_REQUIRES_PREVIOUS_SOURCES_FETCH
@@ -226,12 +240,10 @@ class ElementSources:
                 # CAS-based source cache on its own. Fetch original source
                 # if it's not in the plugin-specific cache yet.
                 if not source._is_cached():
-                    source._fetch(previous_sources)
+                    self._fetch_original_source(source)
             else:
                 self._fetch_source(source)
 
-            previous_sources.append(source)
-
     # get_unique_key():
     #
     # Return something which uniquely identifies the combined sources of the
@@ -246,7 +258,10 @@ class ElementSources:
         result = []
 
         for source in self._sources:
-            result.append({"key": source._get_unique_key(), "name": source.get_kind()})
+            key_dict = {"key": source._get_unique_key(), "name": source.get_kind()}
+            if source._directory:
+                key_dict["directory"] = source._directory
+            result.append(key_dict)
 
         return result
 
@@ -389,34 +404,76 @@ class ElementSources:
 
             # Unable to fetch source from remote source cache, fall back to
             # fetching the original source.
-            source._fetch([])
+            source._fetch()
 
         # Stage original source into the local CAS-based source cache
         self._sourcecache.commit(source)
 
+    # _fetch_source():
+    #
+    # Fetch a single original source
+    #
+    # Args:
+    #   source (Source): The source to fetch
+    #
+    def _fetch_original_source(self, source):
+        if source.BST_REQUIRES_PREVIOUS_SOURCES_FETCH:
+            with self._stage_previous_sources(source) as staging_directory:
+                source._fetch(previous_sources_dir=staging_directory)
+        else:
+            source._fetch()
+
     # _stage():
     #
     # Stage the element sources
     #
-    def _stage(self):
+    # Args:
+    #   stop (Source): Only stage sources listed before this source
+    #
+    def _stage(self, *, stop=None):
         vdir = CasBasedDirectory(self._context.get_cascache())
 
         for source in self._sources:
+            if source == stop:
+                break
+
+            if source._directory:
+                vsubdir = vdir.descend(*source._directory.split(os.sep), create=True)
+            else:
+                vsubdir = vdir
+
             if source.BST_REQUIRES_PREVIOUS_SOURCES_FETCH or source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
                 if source.BST_STAGE_VIRTUAL_DIRECTORY:
-                    source._stage(vdir)
+                    source._stage(vsubdir)
                 else:
                     with utils._tempdir(dir=self._context.tmpdir, prefix="staging-temp") as tmpdir:
                         # Stage previous sources
-                        vdir.export_files(tmpdir)
+                        vsubdir.export_files(tmpdir)
 
                         source._stage(tmpdir)
 
                         # Capture modified tree
-                        vdir._clear()
-                        vdir.import_files(tmpdir)
+                        vsubdir._clear()
+                        vsubdir.import_files(tmpdir)
             else:
                 source_dir = self._sourcecache.export(source)
-                vdir.import_files(source_dir)
+                vsubdir.import_files(source_dir)
 
         return vdir
+
+    # Context manager that stages sources in a cas based or temporary file
+    # based directory
+    @contextmanager
+    def _stage_previous_sources(self, source):
+        self.fetch_sources(stop=source)
+        vdir = self._stage(stop=source)
+
+        if source._directory:
+            vdir = vdir.descend(*source._directory.split(os.sep), create=True)
+
+        if source.BST_STAGE_VIRTUAL_DIRECTORY:
+            yield vdir
+        else:
+            with source.tempdir() as tempdir:
+                vdir.export_files(tempdir)
+                yield tempdir
diff --git a/src/buildstream/source.py b/src/buildstream/source.py
index d7e6021bc..245c3ca99 100644
--- a/src/buildstream/source.py
+++ b/src/buildstream/source.py
@@ -164,7 +164,7 @@ from typing import Iterable, Iterator, Optional, Tuple, TYPE_CHECKING
 from . import _yaml, utils
 from .node import MappingNode
 from .plugin import Plugin
-from .types import SourceRef, Union, List
+from .types import SourceRef, Union
 from ._exceptions import BstError, ImplError, PluginError
 from .exceptions import ErrorDomain
 from ._loader.metasource import MetaSource
@@ -172,7 +172,7 @@ from ._projectrefs import ProjectRefStorage
 from ._cachekey import generate_key
 from .storage import CasBasedDirectory
 from .storage import FileBasedDirectory
-from .storage.directory import Directory, VirtualDirectoryError
+from .storage.directory import Directory
 from ._variables import Variables
 
 if TYPE_CHECKING:
@@ -341,7 +341,7 @@ class Source(Plugin):
 
         self.__element_index = meta.element_index  # The index of the source in the owning element's source list
         self.__element_kind = meta.element_kind  # The kind of the element owning this source
-        self.__directory = meta.directory  # Staging relative directory
+        self._directory = meta.directory  # Staging relative directory
         self.__variables = variables  # The variables used to resolve the source's config
 
         self.__key = None  # Cache key for source
@@ -784,15 +784,11 @@ class Source(Plugin):
     # Wrapper function around plugin provided fetch method
     #
     # Args:
-    #   previous_sources (list): List of Sources listed prior to this source
-    #   fetch_original (bool): whether to fetch full source, or use local CAS
+    #   previous_sources_dir (str): directory where previous sources are staged
     #
-    def _fetch(self, previous_sources):
-
+    def _fetch(self, previous_sources_dir=None):
         if self.BST_REQUIRES_PREVIOUS_SOURCES_FETCH:
-            self.__ensure_previous_sources(previous_sources)
-            with self.__stage_previous_sources(previous_sources) as staging_directory:
-                self.__do_fetch(previous_sources_dir=self.__ensure_directory(staging_directory))
+            self.__do_fetch(previous_sources_dir=previous_sources_dir)
         else:
             self.__do_fetch()
 
@@ -817,8 +813,6 @@ class Source(Plugin):
     # 'directory' option
     #
     def _stage(self, directory):
-        directory = self.__ensure_directory(directory)
-
         if self.BST_KEY_REQUIRES_STAGE:
             # _get_unique_key should be called before _stage
             assert self.__digest is not None
@@ -833,8 +827,6 @@ class Source(Plugin):
         if self.BST_STAGE_VIRTUAL_DIRECTORY:
             directory = FileBasedDirectory(external_directory=directory)
 
-        directory = self.__ensure_directory(directory)
-
         self.validate_cache()
         self.init_workspace(directory)
 
@@ -843,13 +835,10 @@ class Source(Plugin):
     # Wrapper for get_unique_key() api
     #
     def _get_unique_key(self):
-        key = {}
-        key["directory"] = self.__directory
         if self.BST_KEY_REQUIRES_STAGE:
-            key["unique"] = self._stage_into_cas()
+            return self._stage_into_cas()
         else:
-            key["unique"] = self.get_unique_key()  # pylint: disable=assignment-from-no-return
-        return key
+            return self.get_unique_key()
 
     # _project_refs():
     #
@@ -1089,18 +1078,16 @@ class Source(Plugin):
     # Wrapper for track()
     #
     # Args:
-    #   previous_sources (list): List of Sources listed prior to this source
+    #   previous_sources_dir (str): directory where previous sources are staged
     #
-    def _track(self, previous_sources: List["Source"]) -> SourceRef:
+    def _track(self, previous_sources_dir: str = None) -> SourceRef:
         if self.BST_KEY_REQUIRES_STAGE:
             # ensure that these sources have a key after tracking
-            self._get_unique_key()
+            self._generate_key()
             return None
 
         if self.BST_REQUIRES_PREVIOUS_SOURCES_TRACK:
-            self.__ensure_previous_sources(previous_sources)
-            with self.__stage_previous_sources(previous_sources) as staging_directory:
-                new_ref = self.__do_track(previous_sources_dir=self.__ensure_directory(staging_directory))
+            new_ref = self.__do_track(previous_sources_dir=previous_sources_dir)
         else:
             new_ref = self.__do_track()
 
@@ -1116,6 +1103,8 @@ class Source(Plugin):
             # Save ref in local process for subsequent sources
             self._set_ref(new_ref, save=False)
 
+        self._generate_key()
+
         return new_ref
 
     # _is_trackable()
@@ -1206,7 +1195,7 @@ class Source(Plugin):
             self.__element_kind,
             self.get_kind(),
             self.__config,
-            self.__directory,
+            self._directory,
             self.__first_pass,
         )
 
@@ -1221,24 +1210,6 @@ class Source(Plugin):
 
         return clone
 
-    # Context manager that stages sources in a cas based or temporary file
-    # based directory
-    @contextmanager
-    def __stage_previous_sources(self, sources):
-        with self.tempdir() as tempdir:
-            directory = FileBasedDirectory(external_directory=tempdir)
-
-            for src in sources:
-                if src.BST_STAGE_VIRTUAL_DIRECTORY:
-                    src._stage(directory)
-                else:
-                    src._stage(tempdir)
-
-            if self.BST_STAGE_VIRTUAL_DIRECTORY:
-                yield directory
-            else:
-                yield tempdir
-
     # Tries to call fetch for every mirror, stopping once it succeeds
     def __do_fetch(self, **kwargs):
         project = self._get_project()
@@ -1341,31 +1312,6 @@ class Source(Plugin):
             return ref
         raise last_error
 
-    # Ensures a fully constructed path and returns it
-    def __ensure_directory(self, directory):
-
-        if not isinstance(directory, Directory):
-            if self.__directory is not None:
-                directory = os.path.join(directory, self.__directory.lstrip(os.sep))
-
-            try:
-                os.makedirs(directory, exist_ok=True)
-            except OSError as e:
-                raise SourceError(
-                    "Failed to create staging directory: {}".format(e), reason="ensure-stage-dir-fail"
-                ) from e
-
-        else:
-            if self.__directory is not None:
-                try:
-                    directory = directory.descend(*self.__directory.lstrip(os.sep).split(os.sep), create=True)
-                except VirtualDirectoryError as e:
-                    raise SourceError(
-                        "Failed to descend into staging directory: {}".format(e), reason="ensure-stage-dir-fail"
-                    ) from e
-
-        return directory
-
     @classmethod
     def __init_defaults(cls, project, meta):
         if cls.__defaults is None:
@@ -1388,17 +1334,6 @@ class Source(Plugin):
 
         return config
 
-    # Ensures that previous sources have been tracked and fetched.
-    #
-    def __ensure_previous_sources(self, previous_sources):
-        for index, src in enumerate(previous_sources):
-            # BuildStream should track sources in the order they appear so
-            # previous sources should never be in an inconsistent state
-            assert src.is_resolved()
-
-            if not src._is_cached():
-                src._fetch(previous_sources[0:index])
-
 
 def _extract_alias(url):
     parts = url.split(utils._ALIAS_SEPARATOR, 1)