Fix: get rid of dynamic threads

.github/workflows/ci.yaml

@@ -191,7 +191,7 @@ jobs:
          rustup default nightly
          rustup component add rustfmt rust-src
          rustup target add wasm32-unknown-unknown
-         cargo install -f wasm-bindgen-cli --version 0.2.95
+         cargo install -f wasm-bindgen-cli --version 0.2.99
 
      - name: Setup Rust Cache
        uses: Swatinem/rust-cache@v2

core/Cargo.toml

@@ -38,8 +38,8 @@ ark-ff = { workspace = true }
 redux = { workspace = true, features=["serializable_callbacks"] }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
-wasm-bindgen = "0.2"
-wasm-bindgen-futures = "0.4"
+wasm-bindgen = "0.2.99"
+wasm-bindgen-futures = "0.4.49"
 wasm_thread = { version = "0.3", features = [ "es_modules" ] }
 js-sys = "0.3"
 web-sys = { version = "0.3", features = ["Window", "Response"] }

node/common/Cargo.toml

@@ -26,7 +26,7 @@ openmina-core = { path = "../../core" }
 [target.'cfg(target_family = "wasm")'.dependencies]
 redux = { workspace = true }
 wasm-bindgen = "0.2"
-wasm-bindgen-futures = "0.4.42"
+wasm-bindgen-futures = "0.4"
 gloo-timers = { version = "0.3", features = ["futures"] }
 gloo-utils = "0.2"
 tracing-wasm = "0.2"

node/common/src/service/block_producer/mod.rs

@@ -21,18 +21,29 @@ pub struct BlockProducerService {
     provers: Option<BlockProver>,
     keypair: AccountSecretKey,
     vrf_evaluation_sender: mpsc::UnboundedSender<VrfEvaluatorInput>,
+    prove_sender: mpsc::UnboundedSender<(
+        BlockProver,
+        StateHash,
+        Box<ProverExtendBlockchainInputStableV2>,
+    )>,
 }
 
 impl BlockProducerService {
     pub fn new(
         keypair: AccountSecretKey,
         vrf_evaluation_sender: mpsc::UnboundedSender<VrfEvaluatorInput>,
+        prove_sender: mpsc::UnboundedSender<(
+            BlockProver,
+            StateHash,
+            Box<ProverExtendBlockchainInputStableV2>,
+        )>,
         provers: Option<BlockProver>,
     ) -> Self {
         Self {
             provers,
             keypair,
             vrf_evaluation_sender,
+            prove_sender,
         }
     }
 
@@ -41,29 +52,61 @@ impl BlockProducerService {
         keypair: AccountSecretKey,
         provers: Option<BlockProver>,
     ) -> Self {
-        let (vrf_evaluation_sender, vrf_evaluation_receiver) =
-            mpsc::unbounded_channel::<VrfEvaluatorInput>();
+        let (vrf_evaluation_sender, vrf_evaluation_receiver) = mpsc::unbounded_channel();
+        let (prove_sender, prove_receiver) = mpsc::unbounded_channel();
 
+        let event_sender_clone = event_sender.clone();
         let producer_keypair = keypair.clone();
         thread::Builder::new()
             .name("openmina_vrf_evaluator".to_owned())
             .spawn(move || {
                 vrf_evaluator::vrf_evaluator(
-                    event_sender,
+                    event_sender_clone,
                     vrf_evaluation_receiver,
                     producer_keypair.into(),
                 );
             })
             .unwrap();
 
-        BlockProducerService::new(keypair, vrf_evaluation_sender, provers)
+        let producer_keypair = keypair.clone();
+        thread::Builder::new()
+            .name("openmina_block_prover".to_owned())
+            .spawn(move || prover_loop(producer_keypair, event_sender, prove_receiver))
+            .unwrap();
+
+        BlockProducerService::new(keypair, vrf_evaluation_sender, prove_sender, provers)
     }
 
     pub fn keypair(&self) -> AccountSecretKey {
         self.keypair.clone()
     }
 }
 
+fn prover_loop(
+    keypair: AccountSecretKey,
+    event_sender: EventSender,
+    mut rx: mpsc::UnboundedReceiver<(
+        BlockProver,
+        StateHash,
+        Box<ProverExtendBlockchainInputStableV2>,
+    )>,
+) {
+    while let Some((provers, block_hash, input)) = rx.blocking_recv() {
+        let res =
+            prove(provers, input.clone(), keypair.clone(), false).map_err(|err| format!("{err:?}"));
+        if res.is_err() {
+            // IMPORTANT: Make sure that `input` here is a copy from before `prove` is called, we don't
+            // want to leak the private key.
+            if let Err(error) = dump_failed_block_proof_input(block_hash.clone(), input) {
+                openmina_core::error!(
+                        openmina_core::log::system_time();
+                        message = "Failure when dumping failed block proof inputs", error = format!("{error}"));
+            }
+        }
+        let _ = event_sender.send(BlockProducerEvent::BlockProve(block_hash, res).into());
+    }
+}
+
 pub fn prove(
     provers: BlockProver,
     mut input: Box<ProverExtendBlockchainInputStableV2>,
@@ -113,23 +156,12 @@ impl node::service::BlockProducerService for crate::NodeService {
             return;
         }
         let provers = self.provers();
-        let keypair = self.block_producer.as_ref().unwrap().keypair();
-
-        let tx = self.event_sender().clone();
-        thread::spawn(move || {
-            let res =
-                prove(provers, input.clone(), keypair, false).map_err(|err| format!("{err:?}"));
-            if res.is_err() {
-                // IMPORTANT: Make sure that `input` here is a copy from before `prove` is called, we don't
-                // want to leak the private key.
-                if let Err(error) = dump_failed_block_proof_input(block_hash.clone(), input) {
-                    openmina_core::error!(
-                        openmina_core::log::system_time();
-                        message = "Failure when dumping failed block proof inputs", error = format!("{error}"));
-                }
-            }
-            let _ = tx.send(BlockProducerEvent::BlockProve(block_hash, res).into());
-        });
+        let _ = self
+            .block_producer
+            .as_ref()
+            .expect("prove shouldn't be requested if block producer isn't initialized")
+            .prove_sender
+            .send((provers, block_hash, input));
     }
 }
 

node/common/src/service/builder.rs

@@ -125,8 +125,11 @@ impl NodeServiceCommonBuilder {
                 .chain(b"static")
                 .finalize_xof(),
             rng: self.rng,
-            event_sender: self.event_sender,
+            event_sender: self.event_sender.clone(),
             event_receiver: self.event_receiver,
+            snark_block_proof_verify: NodeService::snark_block_proof_verifier_spawn(
+                self.event_sender,
+            ),
             ledger_manager,
             block_producer: self.block_producer,
             p2p,

node/common/src/service/rpc/stats.rs

@@ -23,6 +23,24 @@ impl Stats {
 #[cfg(target_family = "wasm")]
 #[cfg_attr(target_family = "wasm", wasm_bindgen)]
 impl Stats {
+    pub async fn actions(&self, id: JsValue) -> Result<JsValue, JsValue> {
+        let query = if id.is_null() {
+            ActionStatsQuery::SinceStart
+        } else if id.as_string().is_some_and(|s| s == "latest") {
+            ActionStatsQuery::ForLatestBlock
+        } else {
+            let id = id.into_serde().map_err(|err| err.to_string())?;
+            ActionStatsQuery::ForBlockWithId(id)
+        };
+
+        let res = self
+            .sender
+            .oneshot_request::<RpcActionStatsGetResponse>(RpcRequest::ActionStatsGet(query))
+            .await
+            .flatten();
+        Ok(JsValue::from_serde(&res).unwrap_or_default())
+    }
+
     pub async fn sync(&self, limit: Option<usize>) -> JsValue {
         let query = SyncStatsQuery { limit };
         let res = self

node/common/src/service/service.rs

@@ -22,6 +22,7 @@ use super::{
     p2p::webrtc_with_libp2p::P2pServiceCtx,
     replay::ReplayerState,
     rpc::{RpcSender, RpcService},
+    snarks::SnarkBlockVerifyArgs,
     EventReceiver, EventSender,
 };
 
@@ -36,6 +37,8 @@ pub struct NodeService {
     pub event_sender: EventSender,
     pub event_receiver: EventReceiver,
 
+    pub snark_block_proof_verify: mpsc::UnboundedSender<SnarkBlockVerifyArgs>,
+
     pub ledger_manager: LedgerManager,
     pub block_producer: Option<BlockProducerService>,
     pub p2p: P2pServiceCtx,
@@ -105,6 +108,7 @@ impl NodeService {
             rng: StdRng::from_seed(rng_seed),
             event_sender: mpsc::unbounded_channel().0,
             event_receiver: mpsc::unbounded_channel().1.into(),
+            snark_block_proof_verify: mpsc::unbounded_channel().0,
             ledger_manager: LedgerManager::spawn(Default::default()),
             block_producer: None,
             p2p: P2pServiceCtx::mocked(p2p_sec_key),