format code

.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:

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;

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(
+pub async fn put_segment(ctx: &Context, segment: SegmentWithProof) -> RpcResult<()> {
-    ctx: &Context,
-    segment: SegmentWithProof,
-) -> RpcResult<()> {
     debug!(root = %segment.root, index = %segment.index, "putSegment");
 
     // fetch optional tx

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<zgs_grpc_proto::DataRoot> for DataRoot {
     fn try_from(value: zgs_grpc_proto::DataRoot) -> Result<Self, GrpcStatus> {
         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<zgs_grpc_proto::FileProof> 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<zgs_grpc_proto::SegmentWithProof> for SegmentWithProof {
     type Error = GrpcStatus;
 
     fn try_from(grpc_segment: zgs_grpc_proto::SegmentWithProof) -> Result<Self, GrpcStatus> {
-        let root = grpc_segment
+        let root = grpc_segment.root.unwrap().try_into().map_err(|e| e)?;
-            .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
+        let index = grpc_segment.index.try_into().map_err(|_| {
-            .try_into()
+            GrpcStatus::invalid_argument(format!("Invalid segment index: {}", grpc_segment.index))
-            .map_err(|_| GrpcStatus::invalid_argument(format!("Invalid segment index: {}", grpc_segment.index)))?;
+        })?;
-        let proof = grpc_segment
+        let proof = grpc_segment.proof.unwrap().try_into().map_err(|e| e)?;
-            .proof
+        let file_size = grpc_segment.file_size.try_into().map_err(|_| {
-            .unwrap()
+            GrpcStatus::invalid_argument(format!("Invalid file size: {}", grpc_segment.file_size))
-            .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,

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};

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<PingRequest>,
     ) -> Result<tonic::Response<PingReply>, 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<RpcSegment> = segments
+        let rpc_segments = segments
             .into_iter()
-            .map(RpcSegment::try_from)
+            .map(|s| {
-            .collect::<Result<_, _>>()?;
+                RpcSegment::try_from(s)
+                    .map_err(|e| tonic::Status::invalid_argument(format!("Invalid segment: {}", e)))
+            })
+            .collect::<Result<Vec<_>, _>>()?;
 
         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| {
+        let maybe_tx = self
-            tonic::Status::internal(format!("Failed to get transaction by sequence number: {}", e))
+            .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| {
+                .await
-                    tonic::Status::internal(format!("Failed to put segment: {}", e))
+                .map_err(|e| tonic::Status::internal(format!("Failed to put segment: {}", e)))?;
-                })?;
         }
 
         // Return an empty response

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",
         );