workspace.py: Import workspace in get_unique_key

`track` and `fetch` become noop methods and the workspace is imported
into the CAS in the call to `get_unique_key` which also sets the digest
attribute and owns that Directory object. The directory is referenced
during stage to import directly to the virtual directory object.

Importing is expected to be expensive and will be
optimised in future.

When the unique key is retrieved for the workspace source it will also
be commited to the sourcecache. The logic for this source is still a
slight variant on other sources since it cannot itself be expected to
be in the cache when it's opened.

In the source preflight method the preflights of the held sources must
be called.

1 files changed, 53 insertions, 41 deletions

diff --git a/src/buildstream/plugins/sources/workspace.py b/src/buildstream/plugins/sources/workspace.py
index f696410e0..ee145babb 100644
--- a/src/buildstream/plugins/sources/workspace.py
+++ b/src/buildstream/plugins/sources/workspace.py
@@ -37,9 +37,11 @@ workspace. The node constructed would be specified as follows:
 
 import os
 from buildstream.storage.directory import Directory
+from buildstream.storage._casbaseddirectory import CasBasedDirectory
 from buildstream import Source, SourceError, Consistency
 from buildstream import utils
 from buildstream.types import SourceRef
+from buildstream.node import MappingNode
 
 
 class WorkspaceSource(Source):
@@ -47,44 +49,71 @@ class WorkspaceSource(Source):
 
     BST_STAGE_VIRTUAL_DIRECTORY = True
 
-    def __init__(self, context, project, meta):
+    def __init__(self, context, project, meta) -> None:
         super().__init__(context, project, meta)
 
-        # Cached unique key to avoid multiple file system traversal if the unique key is requested multiple times.
+        # Cached unique key
         self.__unique_key = None
+        # the element source objects from the specified metasources
         self.__element_sources = []
-        self.__source_digest = None
+        # the digest of the Directory following the import of the workspace
+        self.__digest = None
+        # the CasBasedDirectory which the path is imported into
+        self.__cas_dir = None
 
-    def set_element_sources(self, _element_sources):
+    def set_element_sources(self, _element_sources: [Source]) -> None:
         self.__element_sources = _element_sources
 
-    def get_element_sources(self):
+    def get_element_sources(self) -> [Source]:
         return self.__element_sources
 
     def track(self) -> SourceRef:
-        return self.__source_digest
+        return None
 
-    def configure(self, node):
+    def configure(self, node: MappingNode) -> None:
         node.validate_keys(['path', 'ref', 'kind'])
         self.path = node.get_str('path')
-        self.__source_digest = node.get_str('ref')
+        self.__digest = node.get_str('ref')
 
-    def preflight(self):
-        return
+    def preflight(self) -> None:
+        for source in self.get_element_sources():
+            source.preflight()
 
     def get_ref(self) -> None:
         return None
 
-    def load_ref(self, node) -> None:
+    def load_ref(self, node: MappingNode) -> None:
         pass  # pragma: nocover
 
-    def set_ref(self, ref, node) -> None:
+    def set_ref(self, ref: SourceRef, node: MappingNode) -> None:
         pass  # pragma: nocover
 
-    def get_unique_key(self):
-        return (self.path, self.__source_digest)
+    def get_unique_key(self) -> (str, SourceRef):
+        sourcecache = self._get_context().sourcecache
+
+        if self.__cas_dir is None:
+            self.__cas_dir = CasBasedDirectory(sourcecache.cas)
+
+        if self.__digest is None:
+
+            with self.timed_activity("Staging local files into CAS"):
+                result = self.__cas_dir.import_files(self.path)
+                if result.overwritten or result.ignored:
+                    raise SourceError(
+                        "Failed to stage source: files clash with existing directory",
+                        reason='ensure-stage-dir-fail')
+                self.__digest = self.__cas_dir._get_digest().hash
+
+        # commit to cache if not cached
+        if not sourcecache.contains(self):
+            sourcecache.commit(self, [])
 
-    def init_workspace(self, directory):
+        #  now close down grpc channels
+        sourcecache.cas.close_channel()
+        assert not sourcecache.cas.has_open_grpc_channels()
+        return (self.path, self.__digest)
+
+    def init_workspace(self, directory: Directory) -> None:
         # for each source held by the workspace we must call init_workspace
         # those sources may override `init_workspace` expecting str or Directory
         # and this will need to be extracted from the directory passed to this method
@@ -94,45 +123,28 @@ class WorkspaceSource(Source):
             source._init_workspace(directory)
 
     def get_consistency(self):
-        if self.__source_digest is None:
-            return Consistency.INCONSISTENT
-        return Consistency.RESOLVED
+        # always return cached state
+        return Consistency.CACHED
 
-    def fetch(self):
+    def fetch(self) -> None:
         pass  # pragma: nocover
 
-    def stage(self, directory):
+    def stage(self, directory: Directory) -> None:
         # directory should always be a Directory object
         assert isinstance(directory, Directory)
-        with self.timed_activity("Staging local files into CAS"):
-            if os.path.isdir(self.path) and not os.path.islink(self.path):
-                result = directory.import_files(self.path)
-            else:
-                result = directory.import_single_file(self.path)
+        assert isinstance(self.__cas_dir, CasBasedDirectory)
+        with self.timed_activity("Staging Workspace files"):
+            result = directory.import_files(self.__cas_dir)
 
             if result.overwritten or result.ignored:
                 raise SourceError(
                     "Failed to stage source: files clash with existing directory",
                     reason='ensure-stage-dir-fail')
 
-    def _get_local_path(self):
+    def _get_local_path(self) -> str:
         return self.path
 
 
-# Create a unique key for a file
-def unique_key(filename):
-
-    # Return some hard coded things for files which
-    # have no content to calculate a key for
-    if os.path.islink(filename):
-        # For a symbolic link, use the link target as its unique identifier
-        return os.readlink(filename)
-    elif os.path.isdir(filename):
-        return "0"
-
-    return utils.sha256sum(filename)
-
-
 # Plugin entry point
-def setup():
+def setup() -> WorkspaceSource:
     return WorkspaceSource