Merge #1014's head 54aae28 (the W3 tree takes the card from plain threads; node pipe) into m4b/spill2

Author: MauroToscano

prover/src/lfm/device_permit.rs

@@ -442,6 +442,19 @@ pub fn hold_labeled(phase: &'static str) -> CardPermit {
             },
         };
     }
+    // ⛔ Never from a rayon worker. A holder that waits inside rayon — a nested
+    // `join` whose other half was stolen, a parallel iterator, another pool's
+    // `install` — runs its pool's queued jobs on its own thread meanwhile, and
+    // one of them that takes the card is a second hold on this thread (BIG 569's
+    // arm N parked there; which run trips it is the scheduler's choice). Taken
+    // from a plain thread, the card is held while rayon builds under it, and
+    // that thread blocks instead of running anyone's jobs.
+    #[cfg(feature = "parallel")]
+    assert!(
+        rayon::current_thread_index().is_none(),
+        "the card permit is taken on a rayon worker ({phase}): a job it runs while it \
+         waits inside rayon could take it again; take it from a plain thread"
+    );
     assert!(
         !HELD_HERE.with(|h| h.get()),
         "the card permit is not reentrant and this thread already holds it; \
@@ -624,6 +637,82 @@ mod tests {
         disarm(&g);
     }
 
+    /// ★ THE RE-ENTRY BIG 569 HIT, FORCED — and now refused at its first hold.
+    /// A holder on a rayon worker that waits on rayon while it holds the card —
+    /// a nested `join` whose other half was stolen, a parallel iterator, another
+    /// pool's `install` — runs queued jobs of its own pool on its own thread
+    /// meanwhile, and one of them that takes the card is a second hold on one
+    /// thread. The W3 tree built its leaves' artifacts as rayon jobs that each
+    /// held the card around a build that uses rayon, so which run tripped it was
+    /// down to the scheduler (568 clean, 569's arm N parked). Here `yield_now` is
+    /// the wait and the sibling the only queued job: before the guard the
+    /// sibling's hold re-entered every time (i-m4b's repro, m4b/permit-reentry);
+    /// with it, the first hold on the worker is refused, before the card is
+    /// taken, so nothing can re-enter.
+    #[cfg(feature = "parallel")]
+    #[test]
+    fn armed_a_hold_on_a_rayon_worker_is_refused_before_a_sibling_can_take_it_again() {
+        let g = ARM.lock().expect("the arm guard is never poisoned");
+        disarm(&g);
+        arm(2);
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(1)
+            .build()
+            .expect("a one-thread pool");
+        let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            pool.install(|| {
+                rayon::join(
+                    || {
+                        let _card = hold_labeled("build_artifacts");
+                        rayon::yield_now()
+                    },
+                    || {
+                        let _card = hold_labeled("build_artifacts");
+                    },
+                )
+            })
+        }));
+        let payload = ran.expect_err("a hold on a rayon worker must be refused");
+        let message = payload
+            .downcast_ref::<String>()
+            .cloned()
+            .or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string()))
+            .unwrap_or_default();
+        assert!(
+            message.contains("the card permit is taken on a rayon worker"),
+            "another panic: {message}"
+        );
+        // Refused before the card was taken: it is free.
+        drop(hold());
+        disarm(&g);
+    }
+
+    /// The other half: armed, a PLAIN thread holds the card while rayon works
+    /// under it (a parallel build, as an artifact build does), and a second
+    /// plain thread's hold waits for it rather than entering beside it.
+    #[cfg(feature = "parallel")]
+    #[test]
+    fn armed_a_plain_thread_holds_the_card_while_rayon_builds_under_it() {
+        use rayon::prelude::*;
+        let g = ARM.lock().expect("the arm guard is never poisoned");
+        disarm(&g);
+        arm(2);
+        let sum = std::thread::scope(|scope| {
+            let card = hold_labeled("build_artifacts");
+            let other = scope.spawn(|| {
+                let _card = hold_labeled("build_artifacts");
+                IN_FLIGHT.load(Ordering::SeqCst)
+            });
+            let sum: u64 = (0..10_000u64).into_par_iter().map(|x| x * x).sum();
+            std::thread::sleep(Duration::from_millis(20));
+            drop(card);
+            assert_eq!(other.join().expect("the second holder"), 1);
+            sum
+        });
+        assert_eq!(sum, (0..10_000u64).map(|x| x * x).sum::<u64>());
+        disarm(&g);
+    }
+
     /// ★ A DEFERRED `multi_prove` WAITS FOR ITS LATCH, AND NOTHING ELSE DOES.
     ///
     /// The worker's artifact hold goes straight to the card; its prove hold