fix: Runaway for unchanged queries participating in cycle (#981)

* fix: Runaway for unchanged queries participating in cycle

* Another regression test

* Fix runaway situation

* Discard changes to src/function/fetch.rs

* Undo tracing changes

* Move accumulated write outside of non-cycle branch

* Short circuit if cycle head is executing

* Inline

* Update expected test output

* Fix double execution

* Simplify check in `validate_same_iteration`

* Some more inline

* Pass references

Author: MichaReiser

src/cycle.rs

@@ -53,6 +53,7 @@ use thin_vec::{thin_vec, ThinVec};
 
 use crate::key::DatabaseKeyIndex;
 use crate::sync::OnceLock;
+use crate::Revision;
 
 /// The maximum number of times we'll fixpoint-iterate before panicking.
 ///
@@ -237,16 +238,30 @@ pub(crate) fn empty_cycle_heads() -> &'static CycleHeads {
 
 #[derive(Debug, PartialEq, Eq)]
 pub enum ProvisionalStatus {
-    Provisional { iteration: IterationCount },
-    Final { iteration: IterationCount },
+    Provisional {
+        iteration: IterationCount,
+        verified_at: Revision,
+    },
+    Final {
+        iteration: IterationCount,
+        verified_at: Revision,
+    },
     FallbackImmediate,
 }
 
 impl ProvisionalStatus {
     pub(crate) const fn iteration(&self) -> Option<IterationCount> {
         match self {
-            ProvisionalStatus::Provisional { iteration } => Some(*iteration),
-            ProvisionalStatus::Final { iteration } => Some(*iteration),
+            ProvisionalStatus::Provisional { iteration, .. } => Some(*iteration),
+            ProvisionalStatus::Final { iteration, .. } => Some(*iteration),
+            ProvisionalStatus::FallbackImmediate => None,
+        }
+    }
+
+    pub(crate) const fn verified_at(&self) -> Option<Revision> {
+        match self {
+            ProvisionalStatus::Provisional { verified_at, .. } => Some(*verified_at),
+            ProvisionalStatus::Final { verified_at, .. } => Some(*verified_at),
             ProvisionalStatus::FallbackImmediate => None,
         }
     }

src/function.rs

@@ -345,10 +345,16 @@ where
             if C::CYCLE_STRATEGY == CycleRecoveryStrategy::FallbackImmediate {
                 ProvisionalStatus::FallbackImmediate
             } else {
-                ProvisionalStatus::Final { iteration }
+                ProvisionalStatus::Final {
+                    iteration,
+                    verified_at: memo.verified_at.load(),
+                }
             }
         } else {
-            ProvisionalStatus::Provisional { iteration }
+            ProvisionalStatus::Provisional {
+                iteration,
+                verified_at: memo.verified_at.load(),
+            }
         })
     }
 

src/function/fetch.rs

@@ -1,3 +1,5 @@
+use rustc_hash::FxHashMap;
+
 use crate::cycle::{CycleHeads, CycleRecoveryStrategy, IterationCount};
 use crate::function::maybe_changed_after::VerifyCycleHeads;
 use crate::function::memo::Memo;
@@ -172,7 +174,6 @@ where
                         zalsa_local,
                         database_key_index,
                         old_memo,
-                        true,
                     )
                 {
                     self.update_shallow(zalsa, database_key_index, old_memo, can_shallow_update);
@@ -182,17 +183,19 @@ where
                     return unsafe { Some(self.extend_memo_lifetime(old_memo)) };
                 }
 
-                let mut cycle_heads = VerifyCycleHeads::default();
+                let mut cycle_heads = Vec::new();
+                let mut participating_queries = FxHashMap::default();
+
                 let verify_result = self.deep_verify_memo(
                     db,
                     zalsa,
                     old_memo,
                     database_key_index,
-                    &mut cycle_heads,
+                    &mut VerifyCycleHeads::new(&mut cycle_heads, &mut participating_queries),
                     can_shallow_update,
                 );
 
-                if verify_result.is_unchanged() && !cycle_heads.has_any() {
+                if verify_result.is_unchanged() && cycle_heads.is_empty() {
                     // SAFETY: memo is present in memo_map and we have verified that it is
                     // still valid for the current revision.
                     return unsafe { Some(self.extend_memo_lifetime(old_memo)) };