Ensure that Tree contents are uploaded after creation. (#13008)

We currently do not upload the file `Digest`s referenced by `Tree`s, which will later cause a "missing file" to be reported during eager fetch of a cache entry.

As discussed in #13006: we should likely have native support for storing and uploading `Tree`s, but this change fixes the immediate issue in a cherry-pickable way.

[ci skip-build-wheels]

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

@@ -496,6 +496,9 @@ impl Store {
   ///
   /// Returns a structure with the summary of operations.
   ///
+  /// TODO: This method is only aware of File and Directory typed blobs: in particular, that means
+  /// it will not expand Trees to upload the files that they refer to. See #13006.
+  ///
   pub fn ensure_remote_has_recursive(
     &self,
     digests: Vec<Digest>,

src/rust/engine/process_execution/src/remote_cache.rs

@@ -115,12 +115,13 @@ impl CommandRunner {
   /// Note that the Tree does not include the directory_path as a prefix, per REAPI. This path
   /// gets stored on the OutputDirectory proto.
   ///
-  /// If the output directory does not exist, then returns Ok(None).
+  /// Returns the created Tree and any File Digests referenced within it. If the output directory
+  /// does not exist, then returns Ok(None).
   pub(crate) async fn make_tree_for_output_directory(
     root_directory_digest: Digest,
     directory_path: RelativePath,
     store: &Store,
-  ) -> Result<Option<Tree>, String> {
+  ) -> Result<Option<(Tree, Vec<Digest>)>, String> {
     // Traverse down from the root directory digest to find the directory digest for
     // the output directory.
     let mut current_directory_digest = root_directory_digest;
@@ -161,7 +162,10 @@ impl CommandRunner {
 
     // At this point, `current_directory_digest` holds the digest of the output directory.
     // This will be the root of the Tree. Add it to a queue of digests to traverse.
+    // TODO: The remainder of this method can be implemented in terms of
+    // `Store::entries_for_directory`, but it does not exist on the 2.7.x branch.
     let mut tree = Tree::default();
+    let mut file_digests = Vec::new();
 
     let mut digest_queue = VecDeque::new();
     digest_queue.push_back(current_directory_digest);
@@ -184,6 +188,15 @@ impl CommandRunner {
         digest_queue.push_back(subdirectory_digest);
       }
 
+      // Collect referenced file Digests.
+      file_digests.extend(
+        directory
+          .files
+          .iter()
+          .map(|file_node| require_digest(file_node.digest.as_ref()))
+          .collect::<Result<Vec<_>, String>>()?,
+      );
+
       // Store this directory either as the `root` or one of the `children` if not the root.
       if directory_digest == current_directory_digest {
         tree.root = Some(directory);
@@ -192,7 +205,7 @@ impl CommandRunner {
       }
     }
 
-    Ok(Some(tree))
+    Ok(Some((tree, file_digests)))
   }
 
   pub(crate) async fn extract_output_file(
@@ -289,19 +302,20 @@ impl CommandRunner {
     digests.insert(result.stderr_digest);
 
     for output_directory in &command.output_directories {
-      let tree = match Self::make_tree_for_output_directory(
+      let (tree, file_digests) = match Self::make_tree_for_output_directory(
         result.output_directory,
         RelativePath::new(output_directory).unwrap(),
         store,
       )
       .await?
       {
-        Some(t) => t,
+        Some(res) => res,
         None => continue,
       };
 
       let tree_digest = crate::remote::store_proto_locally(&self.store, &tree).await?;
       digests.insert(tree_digest);
+      digests.extend(file_digests);
 
       action_result
         .output_directories

src/rust/engine/process_execution/src/remote_cache_tests.rs

@@ -426,7 +426,7 @@ async fn make_tree_from_directory() {
     .await
     .expect("Error saving directory");
 
-  let tree = crate::remote_cache::CommandRunner::make_tree_for_output_directory(
+  let (tree, file_digests) = crate::remote_cache::CommandRunner::make_tree_for_output_directory(
     directory_digest,
     RelativePath::new("pets").unwrap(),
     &store,
@@ -453,6 +453,7 @@ async fn make_tree_from_directory() {
   assert_eq!(file_node.name, "roland.ext");
   let file_digest: Digest = file_node.digest.as_ref().unwrap().try_into().unwrap();
   assert_eq!(file_digest, TestData::roland().digest());
+  assert_eq!(file_digests, vec![TestData::roland().digest()]);
 
   // Test that extracting non-existent output directories fails gracefully.
   assert!(