don't recover when not a participant

When a query Q invokes a cycle Q1...Q1 but Q is not a
participant in that cycle, Q should not recover! Test that.

src/derived/slot.rs

@@ -256,8 +256,14 @@ where
                     crate::plumbing::CycleRecoveryStrategy::Fallback => {
                         if let Some(c) = active_query.take_cycle() {
                             assert!(c.is(&cycle));
+                            Q::cycle_fallback(db, &cycle, &self.key)
+                        } else {
+                            // we are not a participant in this cycle
+                            debug_assert!(!cycle
+                                .participant_keys()
+                                .any(|k| k == self.database_key_index));
+                            cycle.throw()
                         }
-                        Q::cycle_fallback(db, &cycle, &self.key)
                     }
                 }
             }

src/runtime.rs

@@ -7,6 +7,7 @@ use parking_lot::lock_api::{RawRwLock, RawRwLockRecursive};
 use parking_lot::{Mutex, RwLock};
 use rustc_hash::FxHasher;
 use std::hash::{BuildHasherDefault, Hash};
+use std::panic::panic_any;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
@@ -363,13 +364,15 @@ impl Runtime {
 
         self.local_state.restore_query_stack(from_stack);
 
-        if me_recovered || !others_recovered {
-            // if me_recovered: If the current thread has recovery, we want to throw
+        if me_recovered {
+            // If the current thread has recovery, we want to throw
             // so that it can begin.
-            //
-            // otherwise, if !others_recorded: then no threads have recovery, so we want
-            // to throw the cycle so that salsa can abort.
-            cycle.throw();
+            cycle.throw()
+        } else if others_recovered {
+            // If other threads have recovery but we didn't: return and we will block on them.
+        } else {
+            // if nobody has recover, then we panic
+            panic_any(cycle);
         }
     }
 

src/runtime/local_state.rs

@@ -127,8 +127,8 @@ impl LocalState {
                 // They will not have the `cycle` marker set in their
                 // stack frames, so they will just read the fallback value
                 // from `Ci+1` and continue on their merry way.
-                if let Some(cycle) = top_query.cycle.take() {
-                    cycle.throw()
+                if let Some(cycle) = &top_query.cycle {
+                    cycle.clone().throw()
                 }
             }
         })

tests/cycles.rs

@@ -480,3 +480,22 @@ fn cycle_multiple() {
     )
     "###);
 }
+
+#[test]
+fn cycle_recovery_set_but_not_participating() {
+    let mut db = DatabaseImpl::default();
+
+    //     A --> C -+
+    //           ^  |
+    //           +--+
+    db.set_a_invokes(CycleQuery::C);
+    db.set_c_invokes(CycleQuery::C);
+
+    // Here we expect C to panic and A not to recover:
+    let r = extract_cycle(|| drop(db.cycle_a()));
+    insta::assert_debug_snapshot!(r.all_participants(&db), @r###"
+    [
+        "cycle_c(())",
+    ]
+    "###);
+}