Fix out-of-order verification of cycle head dependencies (#1061)

* Start working on maybe_changed_after

* Add `maybe_changed_after` for tracked structs

* Remove old cycle handling in maybe_changed_after

* Add tests

* Don't collect deps if there's an outer cycle

* Remove `validate_provisional`

* Compute outer most cycle

* Set edges in more branches

* More

* Improve logging

* Flatten dependencies of all heads, use lazy validation

* Properly skip `complete_iteration` in the no cycle head case

* Revert `struct_database_key_index` changes

* Revert more changes

* Revert even more

* Rename

* Fix memory leak

* Clippy

* Advoid recursing cycle heads

* Add missing inventory gating

* Only track queries in seen

* Reuse allocations

* Avoid copying current-revision input-outputs into `QueryRevisions`

* Revert "Avoid copying current-revision input-outputs into `QueryRevisions`"

This reverts commit e686db9.

* Move more out of hot path

* Reuse allocations more

* Revert some Ingredient::maybe_changed_after changes

* More cleanup

* Back out disambiguator changes

* Docs

* Small reordering

* Panic when using accumulator in fixpoint query

* Revert unnecessary test changes

* Simplify more

* Code review feedback

* Format

Author: MichaReiser

.github/workflows/test.yml

@@ -124,7 +124,7 @@ jobs:
 
   benchmarks:
     # https://github.com/CodSpeedHQ/action/issues/126
-    if: github.event_name != 'merge_group'
+    if: github.repository == 'salsa-rs/salsa' &&github.event_name != 'merge_group'
     name: Benchmarks
     runs-on: ubuntu-latest
     steps:

src/accumulator.rs

@@ -7,15 +7,15 @@ use std::panic::UnwindSafe;
 
 use accumulated::{Accumulated, AnyAccumulated};
 
-use crate::function::{VerifyCycleHeads, VerifyResult};
+use crate::function::VerifyResult;
 use crate::hash::{FxHashSet, FxIndexSet};
 use crate::ingredient::{Ingredient, Jar};
 use crate::plumbing::ZalsaLocal;
 use crate::sync::Arc;
 use crate::table::memo::MemoTableTypes;
 use crate::zalsa::{IngredientIndex, JarKind, Zalsa};
 use crate::zalsa_local::QueryEdge;
-use crate::{Database, Id, Revision};
+use crate::{Database, DatabaseKeyIndex, Id, Revision};
 
 mod accumulated;
 pub(crate) mod accumulated_map;
@@ -107,7 +107,6 @@ impl<A: Accumulator> Ingredient for IngredientImpl<A> {
         _db: crate::database::RawDatabase<'_>,
         _input: Id,
         _revision: Revision,
-        _cycle_heads: &mut VerifyCycleHeads,
     ) -> VerifyResult {
         panic!("nothing should ever depend on an accumulator directly")
     }
@@ -137,6 +136,16 @@ impl<A: Accumulator> Ingredient for IngredientImpl<A> {
     fn memo_table_types_mut(&mut self) -> &mut Arc<MemoTableTypes> {
         unreachable!("accumulator does not allocate pages")
     }
+
+    fn flatten_cycle_head_dependencies(
+        &self,
+        _zalsa: &Zalsa,
+        _id: Id,
+        _flattened_input_outputs: &mut FxIndexSet<QueryEdge>,
+        _seen: &mut FxHashSet<DatabaseKeyIndex>,
+    ) {
+        panic!("nothing should ever depend on an accumulator directly")
+    }
 }
 
 impl<A> std::fmt::Debug for IngredientImpl<A>

src/active_query.rs

@@ -10,7 +10,9 @@ use crate::key::DatabaseKeyIndex;
 use crate::runtime::Stamp;
 use crate::sync::atomic::AtomicBool;
 use crate::tracked_struct::{Disambiguator, DisambiguatorMap, IdentityHash, IdentityMap};
-use crate::zalsa_local::{QueryEdge, QueryOrigin, QueryRevisions, QueryRevisionsExtra};
+use crate::zalsa_local::{
+    QueryEdge, QueryEdgeKind, QueryOrigin, QueryRevisions, QueryRevisionsExtra,
+};
 use crate::Revision;
 use crate::{
     cycle::{CycleHeads, IterationCount},
@@ -81,10 +83,19 @@ impl ActiveQuery {
     ) {
         assert!(self.input_outputs.is_empty());
 
-        self.input_outputs.extend(edges.iter().cloned());
+        // Copy over outputs for `diff_outputs`, don't copy inputs because cycle heads
+        // flatten all input dependencies.
+        self.input_outputs.extend(
+            edges
+                .iter()
+                .filter(|edge| matches!(edge.kind(), QueryEdgeKind::Output(_)))
+                .copied(),
+        );
         self.durability = self.durability.min(durability);
         self.changed_at = self.changed_at.max(changed_at);
         self.untracked_read |= untracked_read;
+        self.disambiguator_map
+            .seed(active_tracked_ids.iter().map(|(id, _)| id));
 
         // Mark all tracked structs from the previous iteration as active.
         self.tracked_struct_ids

src/cycle.rs

@@ -173,7 +173,7 @@ impl AtomicIterationCount {
         self.0.store(value.0, Ordering::Release);
     }
 
-    pub(crate) fn store_mut(&mut self, value: IterationCount) {
+    pub(crate) fn set(&mut self, value: IterationCount) {
         *self.0.get_mut() = value.0;
     }
 }
@@ -284,7 +284,7 @@ impl CycleHeads {
             .iter_mut()
             .find(|cycle_head| cycle_head.database_key_index == cycle_head_index)
         {
-            cycle_head.iteration_count.store_mut(new_iteration_count);
+            cycle_head.iteration_count.set(new_iteration_count);
         }
     }
 
@@ -329,7 +329,7 @@ impl CycleHeads {
 
             if *removed {
                 *removed = false;
-                existing.iteration_count.store_mut(iteration_count);
+                existing.iteration_count.set(iteration_count);
 
                 true
             } else {
@@ -511,7 +511,7 @@ impl<'db> ProvisionalStatus<'db> {
     pub(crate) fn cycle_heads(&self) -> &'db CycleHeads {
         match self {
             ProvisionalStatus::Provisional { cycle_heads, .. } => cycle_heads,
-            _ => empty_cycle_heads(),
+            ProvisionalStatus::Final { .. } => empty_cycle_heads(),
         }
     }
 

src/function.rs

@@ -1,4 +1,4 @@
-pub(crate) use maybe_changed_after::{VerifyCycleHeads, VerifyResult};
+pub(crate) use maybe_changed_after::VerifyResult;
 pub(crate) use sync::{ClaimGuard, ClaimResult, Reentrancy, SyncGuard, SyncOwner, SyncTable};
 
 use std::any::Any;
@@ -322,11 +322,10 @@ where
         db: RawDatabase<'_>,
         input: Id,
         revision: Revision,
-        cycle_heads: &mut VerifyCycleHeads,
     ) -> VerifyResult {
         // SAFETY: The `db` belongs to the ingredient as per caller invariant
         let db = unsafe { self.view_caster().downcast_unchecked(db) };
-        self.maybe_changed_after(db, input, revision, cycle_heads)
+        self.maybe_changed_after(db, input, revision)
     }
 
     fn collect_minimum_serialized_edges(
@@ -420,6 +419,55 @@ where
         memo.revisions.verified_final.store(true, Ordering::Release);
     }
 
+    fn flatten_cycle_head_dependencies(
+        &self,
+        zalsa: &Zalsa,
+        id: Id,
+        flattened_input_outputs: &mut FxIndexSet<QueryEdge>,
+        seen: &mut FxHashSet<DatabaseKeyIndex>,
+    ) {
+        let memo_index = self.memo_ingredient_index(zalsa, id);
+        let Some(memo) = self.get_memo_from_table_for(zalsa, id, memo_index) else {
+            return;
+        };
+
+        let database_key_index = self.database_key_index(id);
+
+        // Only flatten dependencies of provisional queries, because only those
+        // contain cyclic dependencies.
+        if !memo.may_be_provisional() {
+            flattened_input_outputs.insert(QueryEdge::input(database_key_index));
+            return;
+        }
+
+        // There's nothing to do if we've visited this query before.
+        if !seen.insert(database_key_index) {
+            return;
+        }
+
+        let inputs = memo.revisions.origin.as_ref().inputs();
+
+        match C::CYCLE_STRATEGY {
+            // For queries with cycle handling, simply extend the input/outputs, because
+            // they already flattened their own dependencies when completing the query.
+            CycleRecoveryStrategy::FallbackImmediate | CycleRecoveryStrategy::Fixpoint => {
+                flattened_input_outputs.extend(inputs.map(QueryEdge::input));
+            }
+            // For regular queries, recurse
+            CycleRecoveryStrategy::Panic => {
+                for input in inputs {
+                    let ingredient = zalsa.lookup_ingredient(input.ingredient_index());
+                    ingredient.flatten_cycle_head_dependencies(
+                        zalsa,
+                        input.key_index(),
+                        flattened_input_outputs,
+                        seen,
+                    );
+                }
+            }
+        }
+    }
+
     fn cycle_converged(&self, zalsa: &Zalsa, input: Id) -> bool {
         let Some(memo) =
             self.get_memo_from_table_for(zalsa, input, self.memo_ingredient_index(zalsa, input))

src/function/accumulated.rs

@@ -79,12 +79,7 @@ where
                 stack.reserve(edges.len());
             }
 
-            stack.extend(
-                origin
-                    .inputs()
-                    .filter_map(|input| TryInto::<DatabaseKeyIndex>::try_into(input).ok())
-                    .rev(),
-            );
+            stack.extend(origin.inputs().rev());
 
             visited.reserve(stack.len());
         }

src/function/backdate.rs

@@ -22,7 +22,7 @@ where
         // right now whether backdating could be made safe for queries participating in queries.
         // TODO: Write a test that demonstrates that backdating queries participating in a cycle isn't safe
         // OR write many tests showing that it is (and fixing the case where it didn't correctly account for today).
-        if !revisions.cycle_heads().is_empty() {
+        if !revisions.cycle_heads().is_empty() || old_memo.may_be_provisional() {
             return;
         }