diff --git a/.github/actions/setup-rust/action.yml b/.github/actions/setup-rust/action.yml index b68f39c..632bef4 100644 --- a/.github/actions/setup-rust/action.yml +++ b/.github/actions/setup-rust/action.yml @@ -2,6 +2,12 @@ name: Setup Rust (cache & toolchain) runs: using: composite steps: + - name: Install protoc compiler + shell: bash + run: | + sudo apt-get update + sudo apt-get install -y protobuf-compiler + - name: Install toolchain 1.78.0 uses: actions-rs/toolchain@v1 with: diff --git a/node/rpc/src/lib.rs b/node/rpc/src/lib.rs index bef503a..c34f0a2 100644 --- a/node/rpc/src/lib.rs +++ b/node/rpc/src/lib.rs @@ -8,10 +8,10 @@ mod config; mod error; mod middleware; mod miner; +mod rpc_helper; pub mod types; mod zgs; mod zgs_grpc; -mod rpc_helper; use crate::miner::RpcServer as MinerRpcServer; use crate::types::SegmentWithProof; @@ -31,10 +31,10 @@ use storage_async::Store; use sync::{SyncRequest, SyncResponse, SyncSender}; use task_executor::ShutdownReason; use tokio::sync::broadcast; +use tonic::transport::Server; +use tonic_reflection::server::Builder as ReflectionBuilder; use zgs::RpcServer as ZgsRpcServer; use zgs_miner::MinerMessage; -use tonic_reflection::server::Builder as ReflectionBuilder; -use tonic::transport::Server; pub use admin::RpcClient as ZgsAdminRpcClient; pub use config::Config as RPCConfig; diff --git a/node/rpc/src/rpc_helper.rs b/node/rpc/src/rpc_helper.rs index 4ace1f6..e817898 100644 --- a/node/rpc/src/rpc_helper.rs +++ b/node/rpc/src/rpc_helper.rs @@ -1,15 +1,12 @@ -use crate::Context; -use crate::types::SegmentWithProof; use crate::error; +use crate::types::SegmentWithProof; +use crate::Context; use chunk_pool::SegmentInfo; use jsonrpsee::core::RpcResult; use shared_types::Transaction; /// Put a single segment (mirrors your old `put_segment`) -pub async fn put_segment( - ctx: &Context, - segment: SegmentWithProof, -) -> RpcResult<()> { +pub async fn put_segment(ctx: &Context, segment: SegmentWithProof) -> RpcResult<()> { debug!(root = %segment.root, index = %segment.index, "putSegment"); // fetch optional tx diff --git a/node/rpc/src/types.rs b/node/rpc/src/types.rs index 729ae41..59a2755 100644 --- a/node/rpc/src/types.rs +++ b/node/rpc/src/types.rs @@ -1,5 +1,6 @@ use crate::{error, zgs_grpc_proto}; use append_merkle::ZERO_HASHES; +use ethereum_types::H256 as EthH256; use jsonrpsee::core::RpcResult; use merkle_light::hash::Algorithm; use merkle_light::merkle::{log2_pow2, next_pow2, MerkleTree}; @@ -19,7 +20,6 @@ use storage::config::ShardConfig; use storage::log_store::log_manager::bytes_to_entries; use storage::H256; use tonic::Status as GrpcStatus; -use ethereum_types::H256 as EthH256; const ZERO_HASH: [u8; 32] = [ 0xd3, 0x97, 0xb3, 0xb0, 0x43, 0xd8, 0x7f, 0xcd, 0x6f, 0xad, 0x12, 0x91, 0xff, 0xb, 0xfd, 0x16, @@ -86,7 +86,10 @@ impl TryFrom for DataRoot { fn try_from(value: zgs_grpc_proto::DataRoot) -> Result { let bytes = value.value; if bytes.len() != 32 { - return Err(GrpcStatus::invalid_argument(format!("Invalid hash length: got {}, want 32", bytes.len()))); + return Err(GrpcStatus::invalid_argument(format!( + "Invalid hash length: got {}, want 32", + bytes.len() + ))); } // assume AppDataRoot is a newtype around H256: let mut arr = [0u8; 32]; @@ -104,7 +107,10 @@ impl TryFrom for FileProof { let mut lemma = Vec::with_capacity(value.lemma.len()); for bin in value.lemma { if bin.len() != 32 { - return Err(GrpcStatus::invalid_argument(format!("Invalid hash length: got {}, want 32", bin.len()))); + return Err(GrpcStatus::invalid_argument(format!( + "Invalid hash length: got {}, want 32", + bin.len() + ))); } let mut arr = [0u8; 32]; arr.copy_from_slice(&bin); @@ -123,24 +129,16 @@ impl TryFrom for SegmentWithProof { type Error = GrpcStatus; fn try_from(grpc_segment: zgs_grpc_proto::SegmentWithProof) -> Result { - let root = grpc_segment - .root - .unwrap() - .try_into() - .map_err(|e| e)?; + let root = grpc_segment.root.unwrap().try_into().map_err(|e| e)?; let data = grpc_segment.data; // index is u64 in proto, usize in app - let index = grpc_segment.index - .try_into() - .map_err(|_| GrpcStatus::invalid_argument(format!("Invalid segment index: {}", grpc_segment.index)))?; - let proof = grpc_segment - .proof - .unwrap() - .try_into() - .map_err(|e| e)?; - let file_size = grpc_segment.file_size - .try_into() - .map_err(|_| GrpcStatus::invalid_argument(format!("Invalid file size: {}", grpc_segment.file_size)))?; + let index = grpc_segment.index.try_into().map_err(|_| { + GrpcStatus::invalid_argument(format!("Invalid segment index: {}", grpc_segment.index)) + })?; + let proof = grpc_segment.proof.unwrap().try_into().map_err(|e| e)?; + let file_size = grpc_segment.file_size.try_into().map_err(|_| { + GrpcStatus::invalid_argument(format!("Invalid file size: {}", grpc_segment.file_size)) + })?; Ok(SegmentWithProof { root, diff --git a/node/rpc/src/zgs/impl.rs b/node/rpc/src/zgs/impl.rs index 464ac9d..eb48d33 100644 --- a/node/rpc/src/zgs/impl.rs +++ b/node/rpc/src/zgs/impl.rs @@ -1,7 +1,7 @@ use super::api::RpcServer; -use crate::{error, rpc_helper, SegmentIndexArray}; use crate::types::{FileInfo, Segment, SegmentWithProof, Status}; use crate::Context; +use crate::{error, rpc_helper, SegmentIndexArray}; use jsonrpsee::core::async_trait; use jsonrpsee::core::RpcResult; use shared_types::{DataRoot, FlowProof, Transaction, TxSeqOrRoot, CHUNK_SIZE}; diff --git a/node/rpc/src/zgs_grpc/zgs_grpc.rs b/node/rpc/src/zgs_grpc/zgs_grpc.rs index 54e7c54..e0de902 100644 --- a/node/rpc/src/zgs_grpc/zgs_grpc.rs +++ b/node/rpc/src/zgs_grpc/zgs_grpc.rs @@ -1,7 +1,7 @@ -use crate::zgs_grpc_proto::{PingRequest, PingReply, UploadSegmentsByTxSeqRequest, Empty}; -use crate::zgs_grpc_proto::zgs_grpc_service_server::ZgsGrpcService; -use crate::{rpc_helper, Context, SegmentIndexArray}; use crate::types::SegmentWithProof as RpcSegment; +use crate::zgs_grpc_proto::zgs_grpc_service_server::ZgsGrpcService; +use crate::zgs_grpc_proto::{Empty, PingReply, PingRequest, UploadSegmentsByTxSeqRequest}; +use crate::{rpc_helper, Context, SegmentIndexArray}; pub struct ZgsGrpcServiceImpl { pub ctx: Context, @@ -14,7 +14,9 @@ impl ZgsGrpcService for ZgsGrpcServiceImpl { request: tonic::Request, ) -> Result, tonic::Status> { let msg = request.into_inner().message; - let reply = PingReply { message: format!("Echo: {}", msg) }; + let reply = PingReply { + message: format!("Echo: {}", msg), + }; Ok(tonic::Response::new(reply)) } @@ -26,22 +28,32 @@ impl ZgsGrpcService for ZgsGrpcServiceImpl { let segments = req.segments; let tx_seq = req.tx_seq; - let rpc_segments: Vec = segments + let rpc_segments = segments .into_iter() - .map(RpcSegment::try_from) - .collect::>()?; + .map(|s| { + RpcSegment::try_from(s) + .map_err(|e| tonic::Status::invalid_argument(format!("Invalid segment: {}", e))) + }) + .collect::, _>>()?; let indices = SegmentIndexArray::new(&rpc_segments); info!(%tx_seq, ?indices, "grpc_zgs_uploadSegmentsByTxSeq"); - let maybe_tx = self.ctx.log_store.get_tx_by_seq_number(tx_seq).await.map_err(|e| { - tonic::Status::internal(format!("Failed to get transaction by sequence number: {}", e)) - })?; + let maybe_tx = self + .ctx + .log_store + .get_tx_by_seq_number(tx_seq) + .await + .map_err(|e| { + tonic::Status::internal(format!( + "Failed to get transaction by sequence number: {}", + e + )) + })?; for segment in rpc_segments.into_iter() { rpc_helper::put_segment_with_maybe_tx(&self.ctx, segment, maybe_tx.clone()) - .await.map_err(|e| { - tonic::Status::internal(format!("Failed to put segment: {}", e)) - })?; + .await + .map_err(|e| tonic::Status::internal(format!("Failed to put segment: {}", e)))?; } // Return an empty response diff --git a/node/src/client/builder.rs b/node/src/client/builder.rs index 9610035..34fa28a 100644 --- a/node/src/client/builder.rs +++ b/node/src/client/builder.rs @@ -309,7 +309,9 @@ impl ClientBuilder { executor.spawn( async move { - rpc::run_grpc_server(ctx.clone()).await.expect("Failed to start gRPC server"); + rpc::run_grpc_server(ctx.clone()) + .await + .expect("Failed to start gRPC server"); }, "grpc", );