Two targets can swap positions with pantsd (pantsbuild#7583) (pantsbuild#7617)

Before this PR, nothing would remove the edges of a dirty node, so if
two nodes swapped positions in the graph (e.g. if a dependency between
two targets inverted), a cycle would be detected.

With this PR, if we detect a cycle, but detect that there may be dirty
edges in play, we fully clear that node (including removing its edges),
which will cause it being re-triggered from scratch.

This is specifically in place to handle the cycle scenario - the dirty
bit, and dependency Generations are still the primary mechanism for
handling re-use of old versions.

There's an ugliness here that we still don't remove obsolete edges, so
if Generation 2 of a node has differing dependencies from Generation 1,
the dependency from Generation 1 will still dirty Generation 2. We _may_
want to consider solving that separately as/when it becomes a
significant issue, or we may want to re-work this PR to do something
like that... This PR happens to cover a part of that problem, but only
where it causes definitive problems (a fake cycle) rather than also
where it causes performance problems.

There's probably a slightly more principled solution here along the
lines of:
 * Rather than using () as an edge weight in the graph, use the
   Generation of the dependee Node as an edge weight.
 * When doing cycle detection, compare the edge weight against the
   generation of the node, and ignore obsolete edges.
but I would want to think about that a lot more before doing it...

Author: illicitonion

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

@@ -52,6 +52,52 @@ impl Generation {
   }
 }
 
+///
+/// A result from running a Node.
+///
+/// If the value is Dirty, the consumer should check whether the dependencies of the Node have the
+/// same values as they did when this Node was last run; if so, the value can be re-used
+/// (and should be marked "Clean").
+///
+/// If the value is Clean, the consumer can simply use the value as-is.
+///
+#[derive(Clone, Debug)]
+pub(crate) enum EntryResult<N: Node> {
+  Clean(Result<N::Item, N::Error>),
+  Dirty(Result<N::Item, N::Error>),
+}
+
+impl<N: Node> EntryResult<N> {
+  fn is_dirty(&self) -> bool {
+    if let EntryResult::Dirty(..) = self {
+      true
+    } else {
+      false
+    }
+  }
+
+  fn dirty(&mut self) {
+    if let EntryResult::Clean(value) = self {
+      *self = EntryResult::Dirty(value.clone())
+    }
+  }
+
+  fn clean(&mut self) {
+    if let EntryResult::Dirty(value) = self {
+      *self = EntryResult::Clean(value.clone())
+    }
+  }
+}
+
+impl<N: Node> AsRef<Result<N::Item, N::Error>> for EntryResult<N> {
+  fn as_ref(&self) -> &Result<N::Item, N::Error> {
+    match self {
+      EntryResult::Clean(v) => v,
+      EntryResult::Dirty(v) => v,
+    }
+  }
+}
+
 #[allow(clippy::type_complexity)]
 #[derive(Debug)]
 pub(crate) enum EntryState<N: Node> {
@@ -65,7 +111,7 @@ pub(crate) enum EntryState<N: Node> {
   NotStarted {
     run_token: RunToken,
     generation: Generation,
-    previous_result: Option<Result<N::Item, N::Error>>,
+    previous_result: Option<EntryResult<N>>,
   },
   // A node that is running. A running node that has been marked dirty re-runs rather than
   // completing.
@@ -76,7 +122,7 @@ pub(crate) enum EntryState<N: Node> {
     generation: Generation,
     start_time: Instant,
     waiters: Vec<oneshot::Sender<Result<(N::Item, Generation), N::Error>>>,
-    previous_result: Option<Result<N::Item, N::Error>>,
+    previous_result: Option<EntryResult<N>>,
     dirty: bool,
   },
   // A node that has completed, and then possibly been marked dirty. Because marking a node
@@ -85,9 +131,8 @@ pub(crate) enum EntryState<N: Node> {
   Completed {
     run_token: RunToken,
     generation: Generation,
-    result: Result<N::Item, N::Error>,
+    result: EntryResult<N>,
     dep_generations: Vec<Generation>,
-    dirty: bool,
   },
 }
 
@@ -158,8 +203,9 @@ impl<N: Node> Entry<N> {
     let state = self.state.lock();
     match *state {
       EntryState::Completed {
-        ref result, dirty, ..
-      } if !dirty => Some(result.clone()),
+        result: EntryResult::Clean(ref result),
+        ..
+      } => Some(result.clone()),
       _ => None,
     }
   }
@@ -175,7 +221,7 @@ impl<N: Node> Entry<N> {
     run_token: RunToken,
     generation: Generation,
     previous_dep_generations: Option<Vec<Generation>>,
-    previous_result: Option<Result<N::Item, N::Error>>,
+    previous_result: Option<EntryResult<N>>,
   ) -> EntryState<N>
   where
     C: NodeContext<Node = N>,
@@ -243,16 +289,15 @@ impl<N: Node> Entry<N> {
           start_time: Instant::now(),
           run_token,
           generation,
-          previous_result: previous_result,
+          previous_result,
           dirty: false,
         }
       }
       &EntryKey::Cyclic(_) => EntryState::Completed {
-        result: Err(N::Error::cyclic()),
+        result: EntryResult::Clean(Err(N::Error::cyclic())),
         dep_generations: Vec::new(),
         run_token,
         generation,
-        dirty: false,
       },
     }
   }
@@ -293,10 +338,9 @@ impl<N: Node> Entry<N> {
         &mut EntryState::Completed {
           ref result,
           generation,
-          dirty,
           ..
-        } if !dirty && self.node.content().cacheable() => {
-          return future::result(result.clone())
+        } if self.node.content().cacheable() && !result.is_dirty() => {
+          return future::result(result.as_ref().clone())
             .map(move |res| (res, generation))
             .to_boxed();
         }
@@ -323,21 +367,21 @@ impl<N: Node> Entry<N> {
         EntryState::Completed {
           run_token,
           generation,
-          result,
+          mut result,
           dep_generations,
-          dirty,
         } => {
-          assert!(
-            dirty || !self.node.content().cacheable(),
-            "A clean Node should not reach this point: {:?}",
-            result
-          );
           trace!(
-            "Re-starting node {:?}. It was: dirty={}, cacheable={}",
+            "Re-starting node {:?}. It was: previous_result={:?}, cacheable={}",
             self.node,
-            dirty,
+            result,
             self.node.content().cacheable()
           );
+          assert!(
+            result.is_dirty() || !self.node.content().cacheable(),
+            "A clean Node should not reach this point: {:?}",
+            result
+          );
+          result.dirty();
           // The Node has already completed but is now marked dirty. This indicates that we are the
           // first caller to request it since it was marked dirty. We attempt to clean it (which will
           // cause it to re-run if the dep_generations mismatch).
@@ -416,7 +460,7 @@ impl<N: Node> Entry<N> {
         waiters,
         run_token,
         generation,
-        previous_result,
+        mut previous_result,
         dirty,
         ..
       } => {
@@ -428,6 +472,9 @@ impl<N: Node> Entry<N> {
             "Not completing node {:?} because it was invalidated before completing.",
             self.node
           );
+          if let Some(previous_result) = previous_result.as_mut() {
+            previous_result.dirty();
+          }
           EntryState::NotStarted {
             run_token: run_token.next(),
             generation,
@@ -440,6 +487,9 @@ impl<N: Node> Entry<N> {
             "Not completing node {:?} because it was dirtied before completing.",
             self.node
           );
+          if let Some(previous_result) = previous_result.as_mut() {
+            previous_result.dirty();
+          }
           Self::run(
             context,
             &self.node,
@@ -452,19 +502,19 @@ impl<N: Node> Entry<N> {
         } else {
           // If the new result does not match the previous result, the generation increments.
           let (generation, next_result) = if let Some(result) = result {
-            if Some(&result) == previous_result.as_ref() {
+            if Some(&result) == previous_result.as_ref().map(EntryResult::as_ref) {
               // Node was re-executed, but had the same result value.
-              (generation, result)
+              (generation, EntryResult::Clean(result))
             } else {
-              (generation.next(), result)
+              (generation.next(), EntryResult::Clean(result))
             }
           } else {
             // Node was marked clean.
             // NB: The `expect` here avoids a clone and a comparison: see the method docs.
-            (
-              generation,
-              previous_result.expect("A Node cannot be marked clean without a previous result."),
-            )
+            let mut result =
+              previous_result.expect("A Node cannot be marked clean without a previous result.");
+            result.clean();
+            (generation, result)
           };
           // Notify all waiters (ignoring any that have gone away), and then store the value.
           // A waiter will go away whenever they drop the `Future` `Receiver` of the value, perhaps
@@ -476,14 +526,13 @@ impl<N: Node> Entry<N> {
             waiters.len()
           );
           for waiter in waiters {
-            let _ = waiter.send(next_result.clone().map(|res| (res, generation)));
+            let _ = waiter.send(next_result.as_ref().clone().map(|res| (res, generation)));
           }
           EntryState::Completed {
             result: next_result,
             dep_generations,
             run_token,
             generation,
-            dirty: false,
           }
         }
       }
@@ -541,15 +590,23 @@ impl<N: Node> Entry<N> {
   ///
   /// Clears the state of this Node, forcing it to be recomputed.
   ///
-  pub(crate) fn clear(&mut self) {
+  /// # Arguments
+  ///
+  /// * `graph_still_contains_edges` - If the caller has guaranteed that all edges from this Node
+  ///   have been removed from the graph, they should pass false here, else true. We may want to
+  ///   remove this parameter, and force this method to remove the edges, but that would require
+  ///   acquiring the graph lock here, which we currently don't do.
+  ///
+  pub(crate) fn clear(&mut self, graph_still_contains_edges: bool) {
     let mut state = self.state.lock();
 
-    let (run_token, generation, previous_result) =
+    let (run_token, generation, mut previous_result) =
       match mem::replace(&mut *state, EntryState::initial()) {
         EntryState::NotStarted {
           run_token,
           generation,
           previous_result,
+          ..
         }
         | EntryState::Running {
           run_token,
@@ -567,6 +624,12 @@ impl<N: Node> Entry<N> {
 
     trace!("Clearing node {:?}", self.node);
 
+    if graph_still_contains_edges {
+      if let Some(previous_result) = previous_result.as_mut() {
+        previous_result.dirty();
+      }
+    }
+
     // Swap in a state with a new RunToken value, which invalidates any outstanding work.
     *state = EntryState::NotStarted {
       run_token: run_token.next(),
@@ -585,15 +648,36 @@ impl<N: Node> Entry<N> {
     let state = &mut *self.state.lock();
     trace!("Dirtying node {:?}", self.node);
     match state {
-      &mut EntryState::Running { ref mut dirty, .. }
-      | &mut EntryState::Completed { ref mut dirty, .. } => {
-        // Mark dirty.
+      &mut EntryState::Running { ref mut dirty, .. } => {
         *dirty = true;
       }
+      &mut EntryState::Completed { ref mut result, .. } => {
+        result.dirty();
+      }
       &mut EntryState::NotStarted { .. } => {}
     }
   }
 
+  pub fn may_have_dirty_edges(&self) -> bool {
+    match *self.state.lock() {
+      EntryState::NotStarted {
+        ref previous_result,
+        ..
+      }
+      | EntryState::Running {
+        ref previous_result,
+        ..
+      } => {
+        if let Some(EntryResult::Dirty(..)) = previous_result {
+          true
+        } else {
+          false
+        }
+      }
+      EntryState::Completed { ref result, .. } => result.is_dirty(),
+    }
+  }
+
   pub(crate) fn format(&self) -> String {
     let state = match self.peek() {
       Some(Ok(ref nr)) => format!("{:?}", nr),