Merge 51e91a0 into 1f41051

src/python/pants/engine/process.py

@@ -51,6 +51,7 @@ class Process:
     jdk_home: Optional[str]
     is_nailgunnable: bool
     execution_slot_variable: Optional[str]
+    cache_failures: bool
 
     def __init__(
         self,
@@ -68,6 +69,7 @@ def __init__(
         jdk_home: Optional[str] = None,
         is_nailgunnable: bool = False,
         execution_slot_variable: Optional[str] = None,
+        cache_failures: bool = False,
     ) -> None:
         """Request to run a subprocess, similar to subprocess.Popen.
 
@@ -106,6 +108,7 @@ def __init__(
         self.jdk_home = jdk_home
         self.is_nailgunnable = is_nailgunnable
         self.execution_slot_variable = execution_slot_variable
+        self.cache_failures = cache_failures
 
 
 @frozen_after_init

src/rust/engine/graph/src/entry.rs

@@ -230,6 +230,14 @@ impl<N: Node> Entry<N> {
     &self.node
   }
 
+  pub(crate) fn cacheable_with_output(&self, output: Option<&N::Item>) -> bool {
+    (if let Some(item) = output {
+      self.node.cacheable_item(item)
+    } else {
+      false
+    }) && self.node.cacheable()
+  }
+
   ///
   /// If this Node is currently complete and clean with the given Generation, then waits for it to
   /// be changed in any way. If the node is not clean, or the generation mismatches, returns
@@ -434,7 +442,7 @@ impl<N: Node> Entry<N> {
             entry_id,
             run_token,
             generation,
-            if self.node.cacheable() {
+            if self.cacheable_with_output(Some(result.as_ref())) {
               Some(dep_generations)
             } else {
               None
@@ -506,7 +514,7 @@ impl<N: Node> Entry<N> {
             }
           }
           Some(Ok(result)) => {
-            let next_result: EntryResult<N> = if !self.node.cacheable() {
+            let next_result: EntryResult<N> = if !self.cacheable_with_output(Some(&result)) {
               EntryResult::Uncacheable(result, context.session_id().clone())
             } else if has_weak_deps {
               EntryResult::Dirty(result)

src/rust/engine/graph/src/lib.rs

@@ -863,7 +863,11 @@ impl<N: Node> Graph<N> {
             // This is to allow for the behaviour that an uncacheable Node should always have "dirty"
             // (marked as UncacheableDependencies) dependents, transitively.
             let entry = inner.entry_for_id(edge_ref.target()).unwrap();
-            if !entry.node().cacheable() || entry.has_uncacheable_deps() {
+            let result_item = match result {
+              Some(Ok(ref item)) => Some(item),
+              _ => None,
+            };
+            if !entry.cacheable_with_output(result_item) || entry.has_uncacheable_deps() {
               has_uncacheable_deps = true;
             }
 

src/rust/engine/graph/src/node.rs

@@ -30,9 +30,16 @@ pub trait Node: Clone + Debug + Display + Eq + Hash + Send + 'static {
   async fn run(self, context: Self::Context) -> Result<Self::Item, Self::Error>;
 
   ///
-  /// If the node result is cacheable, return true.
+  /// If a node's output is cacheable based solely on properties of the node, and not the output,
+  /// return true.
   ///
   fn cacheable(&self) -> bool;
+
+  /// A Node may want to compute cacheability differently based on properties of the Node's item.
+  /// The output of this method will be and'd with `cacheable` to compute overall cacheability.
+  fn cacheable_item(&self, _item: &Self::Item) -> bool {
+    self.cacheable()
+  }
 }
 
 pub trait NodeError: Clone + Debug + Eq + Send {

src/rust/engine/process_execution/src/cache.rs

@@ -77,8 +77,10 @@ impl crate::CommandRunner for CommandRunner {
       }
     }
 
+    let cache_failures = req.0.values().any(|process| process.cache_failures);
+
     let result = command_runner.underlying.run(req, context).await?;
-    if result.exit_code == 0 {
+    if result.exit_code == 0 || cache_failures {
       if let Err(err) = command_runner.store(key, &result).await {
         warn!(
           "Error storing process execution result to local cache: {} - ignoring and continuing",

src/rust/engine/process_execution/src/lib.rs

@@ -221,6 +221,8 @@ pub struct Process {
   pub target_platform: PlatformConstraint,
 
   pub is_nailgunnable: bool,
+
+  pub cache_failures: bool,
 }
 
 impl Process {
@@ -250,6 +252,7 @@ impl Process {
       target_platform: PlatformConstraint::None,
       is_nailgunnable: false,
       execution_slot_variable: None,
+      cache_failures: false,
     }
   }
 

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

@@ -65,6 +65,7 @@ fn construct_nailgun_server_request(
     target_platform: platform_constraint,
     is_nailgunnable: true,
     execution_slot_variable: None,
+    cache_failures: false,
   }
 }
 

src/rust/engine/process_execution/src/remote_tests.rs

@@ -90,6 +90,7 @@ async fn make_execute_request() {
     target_platform: PlatformConstraint::None,
     is_nailgunnable: false,
     execution_slot_variable: None,
+    cache_failures: false,
   };
 
   let mut want_command = bazel_protos::remote_execution::Command::new();
@@ -176,6 +177,7 @@ async fn make_execute_request_with_instance_name() {
     target_platform: PlatformConstraint::None,
     is_nailgunnable: false,
     execution_slot_variable: None,
+    cache_failures: false,
   };
 
   let mut want_command = bazel_protos::remote_execution::Command::new();
@@ -275,6 +277,7 @@ async fn make_execute_request_with_cache_key_gen_version() {
     target_platform: PlatformConstraint::None,
     is_nailgunnable: false,
     execution_slot_variable: None,
+    cache_failures: false,
   };
 
   let mut want_command = bazel_protos::remote_execution::Command::new();
@@ -523,6 +526,7 @@ async fn make_execute_request_with_timeout() {
     target_platform: PlatformConstraint::None,
     is_nailgunnable: false,
     execution_slot_variable: None,
+    cache_failures: false,
   };
 
   let mut want_command = bazel_protos::remote_execution::Command::new();

src/rust/engine/process_executor/src/main.rs

@@ -367,6 +367,7 @@ async fn main() {
     target_platform: target_platform,
     is_nailgunnable,
     execution_slot_variable: None,
+    cache_failures: false,
   };
 
   let runner: Box<dyn process_execution::CommandRunner> = match server_arg {

src/rust/engine/src/nodes.rs

@@ -233,7 +233,10 @@ pub fn lift_digest(digest: &Value) -> Result<hashing::Digest, String> {
 /// A Node that represents a set of processes to execute on specific platforms.
 ///
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
-pub struct MultiPlatformExecuteProcess(MultiPlatformProcess);
+pub struct MultiPlatformExecuteProcess {
+  cache_failures: bool,
+  process: MultiPlatformProcess,
+}
 
 impl MultiPlatformExecuteProcess {
   fn lift_execute_process(
@@ -300,6 +303,8 @@ impl MultiPlatformExecuteProcess {
       }
     };
 
+    let cache_failures = externs::project_bool(&value, "cache_failures");
+
     Ok(process_execution::Process {
       argv: externs::project_multi_strs(&value, "argv"),
       env,
@@ -315,6 +320,7 @@ impl MultiPlatformExecuteProcess {
       target_platform,
       is_nailgunnable,
       execution_slot_variable,
+      cache_failures,
     })
   }
 
@@ -341,16 +347,22 @@ impl MultiPlatformExecuteProcess {
       ));
     }
 
+    let mut cache_failures = true;
+
     let mut request_by_constraint: BTreeMap<(PlatformConstraint, PlatformConstraint), Process> =
       BTreeMap::new();
     for (constraint_key, execute_process) in constraint_key_pairs.iter().zip(processes.iter()) {
       let underlying_req =
         MultiPlatformExecuteProcess::lift_execute_process(execute_process, constraint_key.1)?;
+      if !underlying_req.cache_failures {
+        cache_failures = false;
+      }
       request_by_constraint.insert(*constraint_key, underlying_req.clone());
     }
-    Ok(MultiPlatformExecuteProcess(MultiPlatformProcess(
-      request_by_constraint,
-    )))
+    Ok(MultiPlatformExecuteProcess {
+      cache_failures,
+      process: MultiPlatformProcess(request_by_constraint),
+    })
   }
 }
 
@@ -365,7 +377,7 @@ impl WrappedNode for MultiPlatformExecuteProcess {
   type Item = ProcessResult;
 
   async fn run_wrapped_node(self, context: Context) -> NodeResult<ProcessResult> {
-    let request = self.0;
+    let request = self.process;
     let execution_context = process_execution::Context::new(
       context.session.workunit_store(),
       context.session.build_id().to_string(),
@@ -1092,7 +1104,7 @@ impl NodeKey {
   fn workunit_name(&self) -> String {
     match self {
       NodeKey::Task(ref task) => task.task.display_info.name.clone(),
-      NodeKey::MultiPlatformExecuteProcess(mp_epr) => mp_epr.0.workunit_name(),
+      NodeKey::MultiPlatformExecuteProcess(mp_epr) => mp_epr.process.workunit_name(),
       NodeKey::Snapshot(..) => "snapshot".to_string(),
       NodeKey::Paths(..) => "paths".to_string(),
       NodeKey::DigestFile(..) => "digest_file".to_string(),
@@ -1115,7 +1127,8 @@ impl NodeKey {
       NodeKey::Task(ref task) => task.task.display_info.desc.as_ref().map(|s| s.to_owned()),
       NodeKey::Snapshot(ref s) => Some(format!("Snapshotting: {}", s.0)),
       NodeKey::Paths(ref s) => Some(format!("Finding files: {}", s.0)),
-      NodeKey::MultiPlatformExecuteProcess(mp_epr) => Some(mp_epr.0.user_facing_name()),
+      //NodeKey::MultiPlatformExecuteProcess(mp_epr) => Some(mp_epr.0.user_facing_name()),
+      NodeKey::MultiPlatformExecuteProcess(mp_epr) => Some(mp_epr.process.user_facing_name()),
       NodeKey::DigestFile(DigestFile(File { path, .. })) => {
         Some(format!("Fingerprinting: {}", path.display()))
       }
@@ -1289,6 +1302,23 @@ impl Node for NodeKey {
       _ => true,
     }
   }
+
+  fn cacheable_item(&self, output: &NodeOutput) -> bool {
+    match self {
+      NodeKey::MultiPlatformExecuteProcess(ref mp) => match output {
+        NodeOutput::ProcessResult(ref process_result) => {
+          let process_succeeded = process_result.0.exit_code == 0;
+          if mp.cache_failures {
+            true
+          } else {
+            process_succeeded
+          }
+        }
+        _ => true,
+      },
+      _ => true,
+    }
+  }
 }
 
 impl Display for NodeKey {
@@ -1297,7 +1327,7 @@ impl Display for NodeKey {
       &NodeKey::DigestFile(ref s) => write!(f, "DigestFile({})", s.0.path.display()),
       &NodeKey::DownloadedFile(ref s) => write!(f, "DownloadedFile({})", s.0),
       &NodeKey::MultiPlatformExecuteProcess(ref s) => {
-        write!(f, "Process({})", s.0.user_facing_name())
+        write!(f, "Process({})", s.process.user_facing_name())
       }
       &NodeKey::ReadLink(ref s) => write!(f, "ReadLink({})", (s.0).0.display()),
       &NodeKey::Scandir(ref s) => write!(f, "Scandir({})", (s.0).0.display()),