Merge pull request #612 from openmina/feat/node/web

Webnode: expose some rpcs for FE

Author: binier

Cargo.lock

@@ -41,6 +41,9 @@ vergen = { version = "8.2.4", features = ["build", "cargo", "git", "gitcl", "rus
 linkme = { workspace = true }
 redux = { workspace = true, features=["serializable_callbacks"] }
 
+[target.'cfg(target_family = "wasm")'.dependencies]
+wasm-bindgen = "0.2"
+
 [features]
 replay = []
 p2p-webrtc = ["p2p/p2p-webrtc"]

node/Cargo.toml

@@ -23,8 +23,12 @@ node = { path = "../../node", features = ["replay"] }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
 redux = { workspace = true }
+wasm-bindgen = "0.2"
+wasm-bindgen-futures = "0.4.42"
 gloo-timers = { version = "0.3", features = ["futures"] }
+gloo-utils = "0.2"
 tracing-wasm = "0.2"
+
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
 redux = { workspace = true, features=["serializable_callbacks"] }
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }

node/common/Cargo.toml

@@ -1,4 +1,8 @@
-use node::p2p::connection::outgoing::P2pConnectionOutgoingInitOpts;
+mod sender;
+pub use sender::RpcSender;
+
+pub mod stats;
+
 use node::rpc::{
     RpcBlockProducerStatsGetResponse, RpcDiscoveryBoostrapStatsResponse,
     RpcDiscoveryRoutingTableResponse, RpcHealthCheckResponse, RpcLedgerAccountsResponse,
@@ -33,11 +37,6 @@ pub struct NodeRpcRequest {
     pub responder: Box<dyn Send + std::any::Any>,
 }
 
-#[derive(Clone)]
-pub struct RpcSender {
-    tx: mpsc::Sender<NodeRpcRequest>,
-}
-
 pub type RpcReceiver = mpsc::Receiver<NodeRpcRequest>;
 
 pub struct RpcService {
@@ -47,53 +46,6 @@ pub struct RpcService {
     req_receiver: mpsc::Receiver<NodeRpcRequest>,
 }
 
-impl RpcSender {
-    pub fn new(tx: mpsc::Sender<NodeRpcRequest>) -> Self {
-        Self { tx }
-    }
-
-    pub async fn oneshot_request<T>(&self, req: RpcRequest) -> Option<T>
-    where
-        T: 'static + Send + Serialize,
-    {
-        let (tx, rx) = oneshot::channel::<T>();
-        let responder = Box::new(tx);
-        let sender = self.tx.clone();
-        let _ = sender.send(NodeRpcRequest { req, responder }).await;
-
-        rx.await.ok()
-    }
-
-    pub async fn multishot_request<T>(
-        &self,
-        expected_messages: usize,
-        req: RpcRequest,
-    ) -> mpsc::Receiver<T>
-    where
-        T: 'static + Send + Serialize,
-    {
-        let (tx, rx) = mpsc::channel::<T>(expected_messages);
-        let responder = Box::new(tx);
-        let sender = self.tx.clone();
-        let _ = sender.send(NodeRpcRequest { req, responder }).await;
-
-        rx
-    }
-
-    pub async fn peer_connect(
-        &self,
-        opts: P2pConnectionOutgoingInitOpts,
-    ) -> Result<String, String> {
-        let peer_id = opts.peer_id().to_string();
-        let req = RpcRequest::P2pConnectionOutgoing(opts);
-        self.oneshot_request::<RpcP2pConnectionOutgoingResponse>(req)
-            .await
-            .ok_or_else(|| "state machine shut down".to_owned())??;
-
-        Ok(peer_id)
-    }
-}
-
 impl Default for RpcService {
     fn default() -> Self {
         Self::new()

node/common/src/service/rpc.rs node/common/src/service/rpc/mod.rsnode/common/src/service/rpc.rs renamed to node/common/src/service/rpc/mod.rs

@@ -0,0 +1,86 @@
+#[cfg(target_family = "wasm")]
+use gloo_utils::format::JsValueSerdeExt;
+use serde::Serialize;
+#[cfg(target_family = "wasm")]
+use wasm_bindgen::prelude::*;
+
+use node::core::channels::{mpsc, oneshot};
+use node::p2p::connection::outgoing::P2pConnectionOutgoingInitOpts;
+use node::rpc::*;
+
+use super::stats::Stats;
+use super::NodeRpcRequest;
+
+#[derive(Clone)]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+pub struct RpcSender {
+    tx: mpsc::Sender<NodeRpcRequest>,
+}
+
+impl RpcSender {
+    pub fn new(tx: mpsc::Sender<NodeRpcRequest>) -> Self {
+        Self { tx }
+    }
+
+    pub async fn oneshot_request<T>(&self, req: RpcRequest) -> Option<T>
+    where
+        T: 'static + Send + Serialize,
+    {
+        let (tx, rx) = oneshot::channel::<T>();
+        let responder = Box::new(tx);
+        let sender = self.tx.clone();
+        let _ = sender.send(NodeRpcRequest { req, responder }).await;
+
+        rx.await.ok()
+    }
+
+    pub async fn multishot_request<T>(
+        &self,
+        expected_messages: usize,
+        req: RpcRequest,
+    ) -> mpsc::Receiver<T>
+    where
+        T: 'static + Send + Serialize,
+    {
+        let (tx, rx) = mpsc::channel::<T>(expected_messages);
+        let responder = Box::new(tx);
+        let sender = self.tx.clone();
+        let _ = sender.send(NodeRpcRequest { req, responder }).await;
+
+        rx
+    }
+}
+
+impl RpcSender {
+    pub async fn peer_connect(
+        &self,
+        opts: P2pConnectionOutgoingInitOpts,
+    ) -> Result<String, String> {
+        let peer_id = opts.peer_id().to_string();
+        let req = RpcRequest::P2pConnectionOutgoing(opts);
+        self.oneshot_request::<RpcP2pConnectionOutgoingResponse>(req)
+            .await
+            .ok_or_else(|| "state machine shut down".to_owned())??;
+
+        Ok(peer_id)
+    }
+}
+
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+impl RpcSender {
+    pub fn stats(&self) -> Stats {
+        Stats::new(self.clone())
+    }
+}
+
+#[cfg(target_family = "wasm")]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+impl RpcSender {
+    pub async fn status(&self) -> JsValue {
+        let res = self
+            .oneshot_request::<RpcStatusGetResponse>(RpcRequest::StatusGet)
+            .await
+            .flatten();
+        JsValue::from_serde(&res).unwrap_or_default()
+    }
+}

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

@@ -0,0 +1,34 @@
+#[cfg(target_family = "wasm")]
+use gloo_utils::format::JsValueSerdeExt;
+#[cfg(target_family = "wasm")]
+use node::rpc::{RpcBlockProducerStatsGetResponse, RpcRequest};
+#[cfg(target_family = "wasm")]
+use wasm_bindgen::prelude::*;
+
+use super::RpcSender;
+
+#[derive(Clone)]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+pub struct Stats {
+    #[allow(unused)]
+    sender: RpcSender,
+}
+
+impl Stats {
+    pub fn new(sender: RpcSender) -> Self {
+        Self { sender }
+    }
+}
+
+#[cfg(target_family = "wasm")]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+impl Stats {
+    pub async fn block_producer(&self) -> JsValue {
+        let res = self
+            .sender
+            .oneshot_request::<RpcBlockProducerStatsGetResponse>(RpcRequest::BlockProducerStatsGet)
+            .await
+            .flatten();
+        JsValue::from_serde(&res).unwrap_or_default()
+    }
+}

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

@@ -1,5 +1,6 @@
 #![cfg(target_family = "wasm")]
 
+use openmina_node_common::rpc::RpcSender;
 pub use openmina_node_common::*;
 
 mod rayon;
@@ -15,15 +16,24 @@ use wasm_bindgen::prelude::*;
 
 use crate::node::P2pTaskRemoteSpawner;
 
-#[wasm_bindgen]
-pub async fn start() {
-    console_error_panic_hook::set_once();
-    tracing::initialize(tracing::Level::INFO);
+/// Automatically run after wasm is loaded.
+#[wasm_bindgen(start)]
+fn main() {
+    wasm_bindgen_futures::spawn_local(async {
+        console_error_panic_hook::set_once();
+        tracing::initialize(tracing::Level::INFO);
 
-    init_rayon().await.unwrap();
+        init_rayon().await.unwrap();
+    });
+}
 
+#[wasm_bindgen]
+pub async fn run() -> RpcSender {
+    // TODO(binier): what if above init in the `main` function isn't done
+    // and this function gets called.
     let p2p_task_spawner = P2pTaskRemoteSpawner::create();
 
+    let (rpc_sender_tx, rpc_sender_rx) = ::node::core::channels::oneshot::channel();
     let _ = thread::spawn(move || {
         let block_verifier_index = get_verifier_index(VerifierKind::Blockchain).into();
         let work_verifier_index = get_verifier_index(VerifierKind::Transaction).into();
@@ -32,16 +42,19 @@ pub async fn start() {
         node_builder
             .block_verifier_index(block_verifier_index)
             .work_verifier_index(work_verifier_index);
-        node_builder.p2p_no_discovery()
+        node_builder
+            .p2p_no_discovery()
             .p2p_custom_task_spawner(p2p_task_spawner)
             .unwrap();
         node_builder.gather_stats();
         let mut node = node_builder.build().context("node build failed!").unwrap();
+        let _ = rpc_sender_tx.send(node.rpc());
 
         wasm_bindgen_futures::spawn_local(async move {
             node.run_forever().await;
         });
         wasm_bindgen::throw_str("Cursed hack to keep workers alive. See https://github.com/rustwasm/wasm-bindgen/issues/2945");
-    }).join_async().await.unwrap();
-    ::node::core::log::info!(redux::Timestamp::global_now(); "node shut down");
+    });
+
+    rpc_sender_rx.await.unwrap()
 }