Node state for the RPC server

node/tests/src/main.rs

@@ -3,11 +3,8 @@ use std::{fs, io::Write, net::SocketAddr, path::PathBuf, str::FromStr};
 
 use anyhow::{ensure, Context};
 use clap::{Parser, Subcommand};
-use jsonrpsee::{core::client::ClientT, http_client::HttpClientBuilder, rpc_params, types::Params};
-use zksync_consensus_tools::{
-    k8s,
-    rpc::methods::{health_check::HealthCheck, RPCMethod},
-};
+use jsonrpsee::{core::client::ClientT, http_client::HttpClientBuilder, rpc_params};
+use zksync_consensus_tools::{k8s, rpc::methods::health_check};
 
 /// Command line arguments.
 #[derive(Debug, Parser)]
@@ -81,12 +78,11 @@ pub async fn sanity_test() {
         let socket = SocketAddr::from_str(socket).unwrap();
         let url = format!("http://{}", socket);
         let rpc_client = HttpClientBuilder::default().build(url).unwrap();
-        let params = Params::new(None);
         let response: serde_json::Value = rpc_client
-            .request(HealthCheck::method(), rpc_params!())
+            .request(health_check::method(), rpc_params!())
             .await
             .unwrap();
-        assert_eq!(response, HealthCheck::callback(params).unwrap());
+        assert_eq!(response, health_check::callback().unwrap());
     }
 }
 

node/tools/src/bin/deployer.rs

@@ -6,12 +6,10 @@ use std::{fs, path::PathBuf};
 use anyhow::Context;
 use clap::{Parser, Subcommand};
 use zksync_consensus_crypto::{Text, TextFmt};
-use zksync_consensus_roles::{node, validator};
+use zksync_consensus_roles::node::SecretKey;
+use zksync_consensus_roles::validator;
 use zksync_consensus_tools::{k8s, AppConfig, NodeAddr, NODES_PORT};
 
-/// K8s namespace for consensus nodes.
-const NAMESPACE: &str = "consensus";
-
 /// Command line arguments.
 #[derive(Debug, Parser)]
 #[command(name = "deployer")]
@@ -55,7 +53,7 @@ fn generate_config(nodes: usize) -> anyhow::Result<()> {
     // Each node will have `gossip_peers` outbound peers.
     let peers = 2;
 
-    let node_keys: Vec<node::SecretKey> = (0..nodes).map(|_| node::SecretKey::generate()).collect();
+    let node_keys: Vec<SecretKey> = (0..nodes).map(|_| SecretKey::generate()).collect();
 
     let default_config = AppConfig::default_for(setup.genesis.clone());
 
@@ -96,7 +94,7 @@ fn generate_config(nodes: usize) -> anyhow::Result<()> {
 /// Deploys the nodes to the kubernetes cluster.
 async fn deploy(nodes: usize, seed_nodes: Option<usize>) -> anyhow::Result<()> {
     let client = k8s::get_client().await?;
-    k8s::create_or_reuse_namespace(&client, NAMESPACE).await?;
+    k8s::create_or_reuse_namespace(&client, k8s::DEFAULT_NAMESPACE).await?;
 
     let seed_nodes = seed_nodes.unwrap_or(1);
 
@@ -107,13 +105,13 @@ async fn deploy(nodes: usize, seed_nodes: Option<usize>) -> anyhow::Result<()> {
             i,
             true,
             vec![], // Seed peers don't have other peer information
-            NAMESPACE,
+            k8s::DEFAULT_NAMESPACE,
         )
         .await?;
     }
 
     // obtain seed peer(s) IP(s)
-    let peer_ips = k8s::get_seed_node_addrs(&client, seed_nodes, NAMESPACE).await?;
+    let peer_ips = k8s::get_seed_node_addrs(&client, seed_nodes, k8s::DEFAULT_NAMESPACE).await?;
 
     let mut peers = vec![];
 
@@ -129,14 +127,14 @@ async fn deploy(nodes: usize, seed_nodes: Option<usize>) -> anyhow::Result<()> {
 
     // deploy the rest of nodes
     for i in seed_nodes..nodes {
-        k8s::deploy_node(&client, i, false, peers.clone(), NAMESPACE).await?;
+        k8s::deploy_node(&client, i, false, peers.clone(), k8s::DEFAULT_NAMESPACE).await?;
     }
 
     Ok(())
 }
 
 /// Obtain node key from config file.
-fn read_node_key_from_config(node_id: &String) -> anyhow::Result<node::SecretKey> {
+fn read_node_key_from_config(node_id: &String) -> anyhow::Result<SecretKey> {
     let manifest_path = std::env::var("CARGO_MANIFEST_DIR")?;
     let root = PathBuf::from(manifest_path).join("k8s_configs");
     let node_key_path = root.join(node_id).join("node_key");

node/tools/src/main.rs

@@ -124,9 +124,9 @@ async fn main() -> anyhow::Result<()> {
         rpc_addr.set_port(rpc_addr.port() + 100);
     }
 
-    // cloning configuration to let RPCServer show it
-    // TODO this should be queried in real time instead, to reflect any possible change in config
-    let rpc_server = RPCServer::new(rpc_addr, configs.app.clone());
+    // Create the RPC server with the executor's storage.
+    let node_storage = executor.block_store.clone();
+    let rpc_server = RPCServer::new(rpc_addr, node_storage);
 
     // Initialize the storage.
     scope::run!(ctx, |ctx, s| async {

node/tools/src/rpc/methods/health_check.rs

@@ -1,23 +1,16 @@
-//! Health check method for RPC server.
-use super::RPCMethod;
-use jsonrpsee::types::{error::ErrorCode, Params};
+use jsonrpsee::core::RpcResult;
 
-/// Health check method for RPC server.
-pub struct HealthCheck;
-
-impl RPCMethod for HealthCheck {
-    /// Health check response for /health endpoint.
-    fn callback(_params: Params) -> Result<serde_json::Value, ErrorCode> {
-        Ok(serde_json::json!({"health": true}))
-    }
+/// Health check response for /health endpoint.
+pub fn callback() -> RpcResult<serde_json::Value> {
+    Ok(serde_json::json!({"health": true}))
+}
 
-    /// Health check method name.
-    fn method() -> &'static str {
-        "health_check"
-    }
+/// Health check method name.
+pub fn method() -> &'static str {
+    "health_check"
+}
 
-    /// Method path for GET requests.
-    fn path() -> &'static str {
-        "/health"
-    }
+/// Method path for GET requests.
+pub fn path() -> &'static str {
+    "/health"
 }

node/tools/src/rpc/methods/last_commited_block.rs

@@ -0,0 +1,34 @@
+//! Peers method for RPC server.
+use anyhow::Context;
+use jsonrpsee::{
+    core::RpcResult,
+    types::{error::ErrorCode, ErrorObjectOwned},
+};
+use std::sync::Arc;
+use zksync_consensus_storage::BlockStore;
+
+/// Last view response for /last_view endpoint.
+pub fn callback(node_storage: Arc<BlockStore>) -> RpcResult<serde_json::Value> {
+    let sub = &mut node_storage.subscribe();
+    let state = sub.borrow().clone();
+    let last_commited_block_header = state
+        .last
+        .context("Failed to get last state")
+        .map_err(|_| ErrorObjectOwned::from(ErrorCode::InternalError))?
+        .header()
+        .number
+        .0;
+    Ok(serde_json::json!({
+        "last_commited_block": last_commited_block_header
+    }))
+}
+
+/// Last view method name.
+pub fn method() -> &'static str {
+    "last_commited_block"
+}
+
+/// Method path for GET requests.
+pub fn path() -> &'static str {
+    "/last_commited_block"
+}

node/tools/src/rpc/methods/last_view.rs

@@ -0,0 +1,34 @@
+//! Peers method for RPC server.
+use anyhow::Context;
+use jsonrpsee::{
+    core::RpcResult,
+    types::{error::ErrorCode, ErrorObjectOwned},
+};
+use std::sync::Arc;
+use zksync_consensus_storage::BlockStore;
+
+/// Last view response for /last_view endpoint.
+pub fn callback(node_storage: Arc<BlockStore>) -> RpcResult<serde_json::Value> {
+    let sub = &mut node_storage.subscribe();
+    let state = sub.borrow().clone();
+    let last_view = state
+        .last
+        .context("Failed to get last state")
+        .map_err(|_| ErrorObjectOwned::from(ErrorCode::InternalError))?
+        .view()
+        .number
+        .0;
+    Ok(serde_json::json!({
+        "last_view": last_view
+    }))
+}
+
+/// Last view method name.
+pub fn method() -> &'static str {
+    "last_view"
+}
+
+/// Method path for GET requests.
+pub fn path() -> &'static str {
+    "/last_view"
+}

node/tools/src/rpc/methods/mod.rs

@@ -1,15 +1,3 @@
-use jsonrpsee::types::{error::ErrorCode, Params};
-
-/// Trait to implement for new RPC methods.
-pub trait RPCMethod {
-    /// Method response logic when called.
-    fn callback(params: Params) -> Result<serde_json::Value, ErrorCode>;
-    /// Method name.
-    fn method() -> &'static str;
-    /// Method path for GET requests.
-    fn path() -> &'static str;
-}
-
-pub(crate) mod config;
 pub mod health_check;
-pub(crate) mod peers;
+pub mod last_commited_block;
+pub mod last_view;