Use Store::store_file for Snapshot capture and memoized Digest calcul…

…ation via Node.

[ci skip-build-wheels]

src/rust/engine/fs/fs_util/src/main.rs

@@ -385,10 +385,11 @@ async fn execute(top_match: &clap::ArgMatches<'_>) -> Result<(), ExitError> {
             .ok_or_else(|| format!("Tried to save file {:?} but it did not exist", path))?;
           match file {
             fs::Stat::File(f) => {
-              let digest = store::OneOffStoreFileByDigest::new(store.clone(), Arc::new(posix_fs))
-                .store_by_digest(f)
-                .await
-                .unwrap();
+              let digest =
+                store::OneOffStoreFileByDigest::new(store.clone(), Arc::new(posix_fs), false)
+                  .store_by_digest(f)
+                  .await
+                  .unwrap();
 
               let report = ensure_uploaded_to_remote(&store, store_has_remote, digest)
                 .await
@@ -458,7 +459,7 @@ async fn execute(top_match: &clap::ArgMatches<'_>) -> Result<(), ExitError> {
 
         let snapshot = Snapshot::from_path_stats(
           store_copy.clone(),
-          store::OneOffStoreFileByDigest::new(store_copy, posix_fs),
+          store::OneOffStoreFileByDigest::new(store_copy, posix_fs, false),
           paths,
         )
         .await?;

src/rust/engine/fs/src/lib.rs

@@ -39,7 +39,7 @@ pub use crate::glob_matching::{
 };
 
 use std::cmp::min;
-use std::io::{self, Read};
+use std::io;
 use std::ops::Deref;
 use std::os::unix::fs::PermissionsExt;
 use std::path::{Component, Path, PathBuf};
@@ -522,31 +522,8 @@ impl PosixFS {
     self.ignore.is_ignored(stat)
   }
 
-  pub async fn read_file(&self, file: &File) -> Result<FileContent, io::Error> {
-    let path = file.path.clone();
-    let path_abs = self.root.0.join(&file.path);
-    self
-      .executor
-      .spawn_blocking(move || {
-        let is_executable = path_abs.metadata()?.permissions().mode() & 0o100 == 0o100;
-        std::fs::File::open(&path_abs)
-          .and_then(|mut f| {
-            let mut content = Vec::new();
-            f.read_to_end(&mut content)?;
-            Ok(FileContent {
-              path: path,
-              content: Bytes::from(content),
-              is_executable,
-            })
-          })
-          .map_err(|e| {
-            io::Error::new(
-              e.kind(),
-              format!("Failed to read file {:?}: {}", path_abs, e),
-            )
-          })
-      })
-      .await
+  pub fn file_path(&self, file: &File) -> PathBuf {
+    self.root.0.join(&file.path)
   }
 
   pub async fn read_link(&self, link: &Link) -> Result<PathBuf, io::Error> {

src/rust/engine/fs/src/posixfs_tests.rs

@@ -29,37 +29,16 @@ async fn is_executable_true() {
 }
 
 #[tokio::test]
-async fn read_file() {
+async fn file_path() {
   let dir = tempfile::TempDir::new().unwrap();
   let path = PathBuf::from("marmosets");
-  let content = "cute".as_bytes().to_vec();
-  make_file(
-    &std::fs::canonicalize(dir.path()).unwrap().join(&path),
-    &content,
-    0o600,
-  );
   let fs = new_posixfs(&dir.path());
-  let file_content = fs
-    .read_file(&File {
-      path: path.clone(),
-      is_executable: false,
-    })
-    .await
-    .unwrap();
-  assert_eq!(file_content.path, path);
-  assert_eq!(file_content.content, content);
-}
-
-#[tokio::test]
-async fn read_file_missing() {
-  let dir = tempfile::TempDir::new().unwrap();
-  new_posixfs(&dir.path())
-    .read_file(&File {
-      path: PathBuf::from("marmosets"),
-      is_executable: false,
-    })
-    .await
-    .expect_err("Expected error");
+  let expected_path = std::fs::canonicalize(dir.path()).unwrap().join(&path);
+  let actual_path = fs.file_path(&File {
+    path: path.clone(),
+    is_executable: false,
+  });
+  assert_eq!(actual_path, expected_path);
 }
 
 #[tokio::test]

src/rust/engine/fs/store/src/snapshot.rs

@@ -269,7 +269,7 @@ impl Snapshot {
         .map_err(|err| format!("Error expanding globs: {}", err))?;
       Snapshot::from_path_stats(
         store.clone(),
-        OneOffStoreFileByDigest::new(store, posix_fs),
+        OneOffStoreFileByDigest::new(store, posix_fs, true),
         path_stats,
       )
       .await
@@ -354,30 +354,36 @@ pub trait StoreFileByDigest<Error> {
 }
 
 ///
-/// A StoreFileByDigest which reads with a PosixFS and writes to a Store, with no caching.
+/// A StoreFileByDigest which reads immutable files with a PosixFS and writes to a Store, with no
+/// caching.
 ///
 #[derive(Clone)]
 pub struct OneOffStoreFileByDigest {
   store: Store,
   posix_fs: Arc<PosixFS>,
+  immutable: bool,
 }
 
 impl OneOffStoreFileByDigest {
-  pub fn new(store: Store, posix_fs: Arc<PosixFS>) -> OneOffStoreFileByDigest {
-    OneOffStoreFileByDigest { store, posix_fs }
+  pub fn new(store: Store, posix_fs: Arc<PosixFS>, immutable: bool) -> OneOffStoreFileByDigest {
+    OneOffStoreFileByDigest {
+      store,
+      posix_fs,
+      immutable,
+    }
   }
 }
 
 impl StoreFileByDigest<String> for OneOffStoreFileByDigest {
   fn store_by_digest(&self, file: File) -> future::BoxFuture<'static, Result<Digest, String>> {
     let store = self.store.clone();
     let posix_fs = self.posix_fs.clone();
+    let immutable = self.immutable;
     let res = async move {
-      let content = posix_fs
-        .read_file(&file)
+      let path = posix_fs.file_path(&file);
+      store
+        .store_file(true, immutable, move || std::fs::File::open(&path))
         .await
-        .map_err(move |err| format!("Error reading file {:?}: {:?}", file, err))?;
-      store.store_file_bytes(content.content, true).await
     };
     res.boxed()
   }

src/rust/engine/fs/store/src/snapshot_tests.rs

@@ -36,7 +36,7 @@ pub fn setup() -> (
   let dir = tempfile::Builder::new().prefix("root").tempdir().unwrap();
   let ignorer = GitignoreStyleExcludes::create(vec![]).unwrap();
   let posix_fs = Arc::new(PosixFS::new(dir.path(), ignorer, executor).unwrap());
-  let file_saver = OneOffStoreFileByDigest::new(store.clone(), posix_fs.clone());
+  let file_saver = OneOffStoreFileByDigest::new(store.clone(), posix_fs.clone(), true);
   (store, dir, posix_fs, file_saver)
 }
 

src/rust/engine/process_execution/src/local.rs

@@ -117,7 +117,7 @@ impl CommandRunner {
         .await?;
       Snapshot::from_path_stats(
         store.clone(),
-        OneOffStoreFileByDigest::new(store, posix_fs),
+        OneOffStoreFileByDigest::new(store, posix_fs, true),
         path_stats,
       )
       .await

src/rust/engine/src/nodes.rs

@@ -523,16 +523,11 @@ impl WrappedNode for DigestFile {
     context: Context,
     _workunit: &mut RunningWorkunit,
   ) -> NodeResult<hashing::Digest> {
-    let content = context
-      .core
-      .vfs
-      .read_file(&self.0)
-      .map_err(|e| throw(&format!("{}", e)))
-      .await?;
+    let path = context.core.vfs.file_path(&self.0);
     context
       .core
       .store()
-      .store_file_bytes(content.content, true)
+      .store_file(true, false, move || std::fs::File::open(&path))
       .map_err(|e| throw(&e))
       .await
   }