Audit whether MissingDigest errors are propagated.

[ci skip-build-wheels]

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

@@ -483,7 +483,7 @@ pub trait CapturedWorkdir {
     workdir_token: Self::WorkdirToken,
     exclusive_spawn: bool,
     platform: Platform,
-  ) -> Result<FallibleProcessResultWithPlatform, ProcessError> {
+  ) -> Result<FallibleProcessResultWithPlatform, String> {
     let start_time = Instant::now();
 
     // Spawn the process.

src/rust/engine/process_execution/src/nailgun/mod.rs

@@ -149,7 +149,8 @@ impl super::CommandRunner for CommandRunner {
           client_args,
           client_main_class,
           ..
-        } = ParsedJVMCommandLines::parse_command_lines(&req.argv)?;
+        } = ParsedJVMCommandLines::parse_command_lines(&req.argv)
+          .map_err(ProcessError::Unclassified)?;
 
         let nailgun_name = CommandRunner::calculate_nailgun_name(&client_main_class);
         let (client_input_digests, server_input_digests) =
@@ -173,7 +174,7 @@ impl super::CommandRunner for CommandRunner {
             self.inner.immutable_inputs(),
           )
           .await
-          .map_err(|e| format!("Failed to connect to nailgun! {}", e))?;
+          .map_err(|e| e.enrich("Failed to connect to nailgun"))?;
 
         // Prepare the workdir.
         let exclusive_spawn = prepare_workdir(
@@ -204,7 +205,7 @@ impl super::CommandRunner for CommandRunner {
         // release, it assumes that it has been canceled and kills the server.
         nailgun_process.release().await?;
 
-        res
+        Ok(res?)
       }
     )
     .await

src/rust/engine/src/intrinsics.rs

@@ -176,7 +176,7 @@ fn process_request_to_process_result(
 ) -> BoxFuture<'static, NodeResult<Value>> {
   async move {
     let process_request = ExecuteProcess::lift(&context.core.store(), args.pop().unwrap())
-      .map_err(|e| throw(format!("Error lifting Process: {}", e)))
+      .map_err(|e| e.enrich("Error lifting Process"))
       .await?;
 
     let result = context.get(process_request).await?.0;

src/rust/engine/src/nodes.rs

@@ -40,7 +40,7 @@ use crate::externs::engine_aware::{EngineAwareParameter, EngineAwareReturnType};
 use crate::externs::fs::PyFileDigest;
 use graph::{Entry, Node, NodeError, NodeVisualizer};
 use hashing::Digest;
-use store::{self, Store, StoreFileByDigest};
+use store::{self, Store, StoreError, StoreFileByDigest};
 use workunit_store::{
   in_workunit, Level, Metric, ObservationMetric, RunningWorkunit, UserMetadataItem,
   WorkunitMetadata,
@@ -268,7 +268,7 @@ impl ExecuteProcess {
   async fn lift_process_input_digests(
     store: &Store,
     value: &Value,
-  ) -> Result<InputDigests, String> {
+  ) -> Result<InputDigests, StoreError> {
     let input_digests_fut: Result<_, String> = Python::with_gil(|py| {
       let value = (**value).as_ref(py);
       let input_files = lift_directory_digest(externs::getattr(value, "input_digest").unwrap())
@@ -294,10 +294,10 @@ impl ExecuteProcess {
 
     input_digests_fut?
       .await
-      .map_err(|e| format!("Failed to merge input digests for process: {}", e))
+      .map_err(|e| e.enrich("Failed to merge input digests for process"))
   }
 
-  fn lift_process(value: &PyAny, input_digests: InputDigests) -> Result<Process, String> {
+  fn lift_process(value: &PyAny, input_digests: InputDigests) -> Result<Process, StoreError> {
     let env = externs::getattr_from_str_frozendict(value, "env");
     let working_directory = match externs::getattr_as_optional_string(value, "working_directory") {
       None => None,
@@ -374,7 +374,7 @@ impl ExecuteProcess {
     })
   }
 
-  pub async fn lift(store: &Store, value: Value) -> Result<Self, String> {
+  pub async fn lift(store: &Store, value: Value) -> Result<Self, StoreError> {
     let input_digests = Self::lift_process_input_digests(store, &value).await?;
     let process = Python::with_gil(|py| Self::lift_process((*value).as_ref(py), input_digests))?;
     Ok(Self { process })