Fix cycle error

Author: MichaReiser

src/function/fetch.rs

@@ -179,24 +179,53 @@ where
         };
 
         // Now that we've claimed the item, check again to see if there's a "hot" value.
-        let opt_old_memo = self.get_memo_from_table_for(zalsa, id, memo_ingredient_index);
+        let mut opt_old_memo = self.get_memo_from_table_for(zalsa, id, memo_ingredient_index);
+
+        // If this is a provisional memo from the same revision. Await all cycle heads because they could be
+        // running on a different thread.
+        if let Some(mut old_memo) = opt_old_memo {
+            if old_memo.value.is_some() {
+                if old_memo.may_be_provisional()
+                    && old_memo.verified_at.load() == zalsa.current_revision()
+                {
+                    opt_old_memo = loop {
+                        old_memo.await_heads(zalsa, self.database_key_index(id));
+
+                        let new_old =
+                            self.get_memo_from_table_for(zalsa, id, memo_ingredient_index);
+                        match new_old {
+                            None => unreachable!("Expected memo to be present"),
+                            // If the new memo is the same as the old, then this means that this is still the "latest" memo for this
+                            Some(new_old) if std::ptr::eq(new_old, old_memo) => {
+                                break Some(new_old);
+                            }
+                            Some(new_old) => {
+                                tracing::debug!("Memo has been updated while waiting for cycle heads");
+                                old_memo = new_old;
+                            }
+                        }
+                    };
+                }
+            }
+        }
+
         if let Some(old_memo) = opt_old_memo {
             if old_memo.value.is_some() {
-                let mut cycle_heads = CycleHeads::default();
-                if let VerifyResult::Unchanged(_) = self.deep_verify_memo(
-                    db,
-                    zalsa,
-                    old_memo,
-                    self.database_key_index(id),
-                    &mut cycle_heads,
-                ) {
-                    if 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)) };
-                    }
+            let mut cycle_heads = CycleHeads::default();
+            if let VerifyResult::Unchanged(_) = self.deep_verify_memo(
+                db,
+                zalsa,
+                old_memo,
+                self.database_key_index(id),
+                &mut cycle_heads,
+            ) {
+                if 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)) };
                 }
             }
+                }
         }
 
         let memo = self.execute(

src/function/memo.rs

@@ -155,7 +155,6 @@ impl<V> Memo<V> {
             for head in cycle_heads {
                 let head_index = head.database_key_index;
 
-
                 let ingredient = zalsa.lookup_ingredient(head_index.ingredient_index());
                 let cycle_head_kind = ingredient.cycle_head_kind(zalsa, head_index.key_index());
                 if matches!(
@@ -200,6 +199,37 @@ impl<V> Memo<V> {
         }
     }
 
+    #[inline(always)]
+    pub(super) fn await_heads(&self, zalsa: &Zalsa, database_key_index: DatabaseKeyIndex) {
+        for head in &self.revisions.cycle_heads {
+            let head_index = head.database_key_index;
+
+            if database_key_index == head_index {
+                continue;
+            }
+
+            let ingredient = zalsa.lookup_ingredient(head_index.ingredient_index());
+            let cycle_head_kind = ingredient.cycle_head_kind(zalsa, head_index.key_index());
+
+            if matches!(
+                cycle_head_kind,
+                CycleHeadKind::NotProvisional | CycleHeadKind::FallbackImmediate
+            ) {
+                // This cycle is already finalized, so we don't need to wait on it;
+                // keep looping through cycle heads.
+                tracing::trace!("Dependent cycle head {head_index:?} has been finalized.");
+            } else if ingredient.wait_for(zalsa, head_index.key_index()) {
+                tracing::trace!(
+                    "Dependent cycle head {head_index:?} has been released"
+                );
+            } else {
+                // We hit a cycle blocking on the cycle head; this means it's in
+                // our own active query stack and we are responsible to resolve the
+                // cycle
+            }
+        }
+    }
+
     /// Cycle heads that should be propagated to dependent queries.
     #[inline(always)]
     pub(super) fn cycle_heads(&self) -> &CycleHeads {