Build the prover without parallel again

`make compile-recursion-elfs` compiles `lambda-vm-prover` for the RISC-V guest,
where `parallel` is off and there is no rayon. The three new phase modules
`use rayon::prelude::*` unconditionally and call `into_par_iter`/`par_iter`, so
the recursion guest stopped building: 27 errors, 8 unresolved-`rayon` and 9
missing-method, plus three `E0505`s in `trace_builder`.

This is on #994's branch as it stands, not introduced by this PR — the same
`cargo check -p lambda-vm-prover --no-default-features` fails identically at
`f800e4b0`. It went unnoticed because no CI run has ever touched that branch;
this PR is the first, which is how it surfaced. The four `make lint` arms do
not catch it either: the workspace-level `--no-default-features` arm still
resolves `parallel` through another member's feature unification.

Gated with the idiom already used in `trace_builder.rs` — a `#[cfg]` pair
around the iterator source, serial arm `into_iter`/`iter`. Where the closure
was long enough that duplicating it would be worse than the problem, it is
hoisted to a named binding first and both arms map over that, so the body
appears once. No behaviour change on any path that runs today: the serial arms
exist to compile for the guest, which links the crate for its verifier and
never executes these phases.

The `E0505`s were the serial arm of the BITWISE collector loop iterating
`&collectors` where the parallel arm moves it into `units`, so the closures'
borrows of the op lists outlived the point where `CollectedOps` moves those
lists. Consumed by value, matching the parallel arm.

Verified: `make compile-recursion-elfs` succeeds, all four `make lint` arms and
`cargo fmt --check` pass, and the prove-and-retire tests are unchanged at 13/13.

Author: MauroToscano

prover/src/challenge_phase.rs

@@ -17,6 +17,7 @@
 
 use std::collections::HashMap;
 
+#[cfg(feature = "parallel")]
 use rayon::prelude::*;
 
 use crypto::fiat_shamir::default_transcript::DefaultTranscript;
@@ -159,12 +160,17 @@ fn assemble_roots(
         (&airs.register, &remaining.register, "REGISTER"),
     ];
     // Small tables, many of them: one commit at a time leaves most cores idle.
-    roots.extend(
-        fixed
-            .par_iter()
-            .map(|(air, trace, name)| commit_resident(air, trace, name))
-            .collect::<Result<Vec<_>, _>>()?,
-    );
+    #[cfg(feature = "parallel")]
+    let fixed_roots = fixed
+        .par_iter()
+        .map(|(air, trace, name)| commit_resident(air, trace, name))
+        .collect::<Result<Vec<_>, _>>()?;
+    #[cfg(not(feature = "parallel"))]
+    let fixed_roots = fixed
+        .iter()
+        .map(|(air, trace, name)| commit_resident(air, trace, name))
+        .collect::<Result<Vec<_>, _>>()?;
+    roots.extend(fixed_roots);
     if airs.include_halt {
         roots.push(commit_resident(&airs.halt, &remaining.halt, "HALT")?);
     }
@@ -184,12 +190,17 @@ fn assemble_roots(
             // PAGE is built from the ELF image rather than from an op list, so
             // it is never retired and is committed here with the rest.
             let pages: Vec<_> = page_airs.by_ref().collect();
-            roots.extend(
-                pages
-                    .par_iter()
-                    .map(|(air, trace)| commit_resident(air, trace, "PAGE"))
-                    .collect::<Result<Vec<_>, _>>()?,
-            );
+            #[cfg(feature = "parallel")]
+            let page_roots = pages
+                .par_iter()
+                .map(|(air, trace)| commit_resident(air, trace, "PAGE"))
+                .collect::<Result<Vec<_>, _>>()?;
+            #[cfg(not(feature = "parallel"))]
+            let page_roots = pages
+                .iter()
+                .map(|(air, trace)| commit_resident(air, trace, "PAGE"))
+                .collect::<Result<Vec<_>, _>>()?;
+            roots.extend(page_roots);
             continue;
         };
         for chunk in 0..count_for(table_counts, kind) {

prover/src/commit_phase.rs

@@ -100,18 +100,20 @@ fn commit_batch(
     roots: &std::sync::Mutex<Vec<ChunkCommitment>>,
     items: Vec<Item>,
 ) -> Result<(), Error> {
+    #[cfg(feature = "parallel")]
     use rayon::prelude::*;
     type P = stark::prover::Prover<GoldilocksField, GoldilocksExtension, ()>;
-    let done: Result<Vec<ChunkCommitment>, Error> = items
-        .into_par_iter()
-        .map(|(kind, chunk, trace)| {
-            <P as IsStarkProver<_, _, _>>::commit_table_root(airs.get(kind).as_ref(), &trace)
-                .map(|root| (kind, chunk, root))
-                .ok_or_else(|| {
-                    Error::Prover(format!("commit phase: no commitment for a {kind:?} chunk"))
-                })
-        })
-        .collect();
+    let commit = |(kind, chunk, trace): Item| -> Result<ChunkCommitment, Error> {
+        <P as IsStarkProver<_, _, _>>::commit_table_root(airs.get(kind).as_ref(), &trace)
+            .map(|root| (kind, chunk, root))
+            .ok_or_else(|| {
+                Error::Prover(format!("commit phase: no commitment for a {kind:?} chunk"))
+            })
+    };
+    #[cfg(feature = "parallel")]
+    let done: Result<Vec<ChunkCommitment>, Error> = items.into_par_iter().map(commit).collect();
+    #[cfg(not(feature = "parallel"))]
+    let done: Result<Vec<ChunkCommitment>, Error> = items.into_iter().map(commit).collect();
     roots.lock().expect("roots").extend(done?);
     Ok(())
 }

prover/src/logup_phase.rs

@@ -97,28 +97,30 @@ fn rounds_batch(
     done: &Proofs,
     items: Vec<Item>,
 ) -> Result<(), Error> {
+    #[cfg(feature = "parallel")]
     use rayon::prelude::*;
     let order = &challenge.order;
     let n = order.len();
-    let built: Result<Vec<_>, Error> = items
-        .into_par_iter()
-        .map(|(kind, chunk, mut trace)| {
-            let idx = order.index_of(kind, chunk).ok_or_else(|| {
-                Error::Prover(format!(
-                    "logup phase: {kind:?} chunk {chunk} is not in the layout the Commit phase produced"
-                ))
-            })?;
-            let mut transcript = fork(&challenge.transcript, idx, n);
-            let rounds = prove_table(
-                airs.get(kind).as_ref(),
-                &mut trace,
-                &challenge.challenges,
-                &mut transcript,
-            )
-            .map_err(|e| Error::Prover(format!("logup phase: {kind:?} chunk {chunk}: {e}")))?;
-            Ok((idx, rounds))
-        })
-        .collect();
+    let run_one = |(kind, chunk, mut trace): Item| {
+        let idx = order.index_of(kind, chunk).ok_or_else(|| {
+            Error::Prover(format!(
+                "logup phase: {kind:?} chunk {chunk} is not in the layout the Commit phase produced"
+            ))
+        })?;
+        let mut transcript = fork(&challenge.transcript, idx, n);
+        let rounds = prove_table(
+            airs.get(kind).as_ref(),
+            &mut trace,
+            &challenge.challenges,
+            &mut transcript,
+        )
+        .map_err(|e| Error::Prover(format!("logup phase: {kind:?} chunk {chunk}: {e}")))?;
+        Ok((idx, rounds))
+    };
+    #[cfg(feature = "parallel")]
+    let built: Result<Vec<_>, Error> = items.into_par_iter().map(run_one).collect();
+    #[cfg(not(feature = "parallel"))]
+    let built: Result<Vec<_>, Error> = items.into_iter().map(run_one).collect();
     done.lock().expect("logup results").extend(built?);
     Ok(())
 }
@@ -235,6 +237,7 @@ fn assemble(
     resident: &mut Resident,
     challenge: &Challenge,
 ) -> Result<Vec<StarkProof<GoldilocksField, GoldilocksExtension, ()>>, Error> {
+    #[cfg(feature = "parallel")]
     use rayon::prelude::*;
     let order = &challenge.order;
     let airs = &challenge.airs;
@@ -299,10 +302,16 @@ fn assemble(
             .ok_or_else(|| Error::Prover(format!("logup phase: page {i} is not in the layout")))?;
         jobs.push((idx, air, trace));
     }
+    #[cfg(feature = "parallel")]
     let built: Vec<(usize, StarkProof<GoldilocksField, GoldilocksExtension, ()>)> = jobs
         .into_par_iter()
         .map(|(idx, air, trace)| build(idx, air, trace).map(|proof| (idx, proof)))
         .collect::<Result<_, _>>()?;
+    #[cfg(not(feature = "parallel"))]
+    let built: Vec<(usize, StarkProof<GoldilocksField, GoldilocksExtension, ()>)> = jobs
+        .into_iter()
+        .map(|(idx, air, trace)| build(idx, air, trace).map(|proof| (idx, proof)))
+        .collect::<Result<_, _>>()?;
     for (idx, proof) in built {
         slots[idx] = Some(proof);
     }
@@ -409,29 +418,31 @@ fn deep_batch(
     done: &Deeps,
     items: Vec<Item>,
 ) -> Result<(), Error> {
+    #[cfg(feature = "parallel")]
     use rayon::prelude::*;
     let order = &challenge.order;
     let n = order.len();
-    let built: Result<Vec<_>, Error> = items
-        .into_par_iter()
-        .map(|(kind, chunk, mut trace)| {
-            let idx = order.index_of(kind, chunk).ok_or_else(|| {
-                Error::Prover(format!(
-                    "batched phase: {kind:?} chunk {chunk} is not in the layout"
-                ))
-            })?;
-            let mut transcript = fork(&challenge.transcript, idx, n);
-            let deep = deep_of(
-                airs.get(kind).as_ref(),
-                &mut trace,
-                &challenge.challenges,
-                &mut transcript,
-                challenge.roots.get(idx).cloned(),
-            )
-            .map_err(|e| Error::Prover(format!("batched phase: {kind:?} chunk {chunk}: {e}")))?;
-            Ok((idx, deep))
-        })
-        .collect();
+    let run_one = |(kind, chunk, mut trace): Item| {
+        let idx = order.index_of(kind, chunk).ok_or_else(|| {
+            Error::Prover(format!(
+                "batched phase: {kind:?} chunk {chunk} is not in the layout"
+            ))
+        })?;
+        let mut transcript = fork(&challenge.transcript, idx, n);
+        let deep = deep_of(
+            airs.get(kind).as_ref(),
+            &mut trace,
+            &challenge.challenges,
+            &mut transcript,
+            challenge.roots.get(idx).cloned(),
+        )
+        .map_err(|e| Error::Prover(format!("batched phase: {kind:?} chunk {chunk}: {e}")))?;
+        Ok((idx, deep))
+    };
+    #[cfg(feature = "parallel")]
+    let built: Result<Vec<_>, Error> = items.into_par_iter().map(run_one).collect();
+    #[cfg(not(feature = "parallel"))]
+    let built: Result<Vec<_>, Error> = items.into_iter().map(run_one).collect();
     // Folded in a fixed order within the batch, so the sequence is a function
     // of the walk and not of which thread finished first.
     let mut built = built?;
@@ -548,6 +559,7 @@ pub fn run_batched(
     let airs = &challenge.airs;
     let n = order.len();
     {
+        #[cfg(feature = "parallel")]
         use rayon::prelude::*;
         let mut state = done.lock().expect("fold state");
         let build = |idx: usize,
@@ -600,10 +612,16 @@ pub fn run_batched(
             })?;
             jobs.push((idx, air, trace));
         }
+        #[cfg(feature = "parallel")]
         let deeps: Vec<(usize, Deep)> = jobs
             .into_par_iter()
             .map(|(idx, air, trace)| build(idx, air, trace))
             .collect::<Result<_, _>>()?;
+        #[cfg(not(feature = "parallel"))]
+        let deeps: Vec<(usize, Deep)> = jobs
+            .into_iter()
+            .map(|(idx, air, trace)| build(idx, air, trace))
+            .collect::<Result<_, _>>()?;
         for (idx, deep) in deeps {
             fold_one(&mut state, idx, deep);
         }
@@ -714,30 +732,32 @@ fn open_batch(
     done: &Opens,
     items: Vec<Item>,
 ) -> Result<(), Error> {
+    #[cfg(feature = "parallel")]
     use rayon::prelude::*;
     let order = &challenge.order;
     let n = order.len();
-    let built: Result<Vec<_>, Error> = items
-        .into_par_iter()
-        .map(|(kind, chunk, mut trace)| {
-            let idx = order.index_of(kind, chunk).ok_or_else(|| {
-                Error::Prover(format!(
-                    "open pass: {kind:?} chunk {chunk} is not in the layout"
-                ))
-            })?;
-            let mut transcript = fork(&challenge.transcript, idx, n);
-            let opening = open_of(
-                airs.get(kind).as_ref(),
-                &mut trace,
-                &challenge.challenges,
-                &mut transcript,
-                iotas_of(batched, idx)?,
-                take_kept(batched, idx),
-            )
-            .map_err(|e| Error::Prover(format!("open pass: {kind:?} chunk {chunk}: {e}")))?;
-            Ok((idx, opening))
-        })
-        .collect();
+    let run_one = |(kind, chunk, mut trace): Item| {
+        let idx = order.index_of(kind, chunk).ok_or_else(|| {
+            Error::Prover(format!(
+                "open pass: {kind:?} chunk {chunk} is not in the layout"
+            ))
+        })?;
+        let mut transcript = fork(&challenge.transcript, idx, n);
+        let opening = open_of(
+            airs.get(kind).as_ref(),
+            &mut trace,
+            &challenge.challenges,
+            &mut transcript,
+            iotas_of(batched, idx)?,
+            take_kept(batched, idx),
+        )
+        .map_err(|e| Error::Prover(format!("open pass: {kind:?} chunk {chunk}: {e}")))?;
+        Ok((idx, opening))
+    };
+    #[cfg(feature = "parallel")]
+    let built: Result<Vec<_>, Error> = items.into_par_iter().map(run_one).collect();
+    #[cfg(not(feature = "parallel"))]
+    let built: Result<Vec<_>, Error> = items.into_iter().map(run_one).collect();
     done.lock().expect("openings").extend(built?);
     Ok(())
 }
@@ -868,12 +888,19 @@ pub fn run_open(
         jobs.push((idx, air, trace));
     }
     {
+        #[cfg(feature = "parallel")]
         use rayon::prelude::*;
-        opens.extend(
-            jobs.into_par_iter()
-                .map(|(idx, air, trace)| build(idx, air, trace))
-                .collect::<Result<Vec<_>, _>>()?,
-        );
+        #[cfg(feature = "parallel")]
+        let opened = jobs
+            .into_par_iter()
+            .map(|(idx, air, trace)| build(idx, air, trace))
+            .collect::<Result<Vec<_>, _>>()?;
+        #[cfg(not(feature = "parallel"))]
+        let opened = jobs
+            .into_iter()
+            .map(|(idx, air, trace)| build(idx, air, trace))
+            .collect::<Result<Vec<_>, _>>()?;
+        opens.extend(opened);
     }
 
     opens.sort_by_key(|(idx, _)| *idx);

prover/src/tables/trace_builder.rs

@@ -4226,7 +4226,10 @@ fn build_traces<I: ImageSource + Sync>(
     {
         base.add_ops(&bitwise_ops);
         memw_register::collect_bitwise_from_memw_register(&memw_register_rows, &mut base);
-        for f in &collectors {
+        // By value, like the parallel arm's `units.extend(collectors)`: these
+        // closures borrow the op lists, and the lists are moved into
+        // `CollectedOps` below, so the collectors have to be dropped here.
+        for f in collectors {
             f(&mut base);
         }
     }