modify streaming update client

Author: yoshidan

modules/service/src/elc.rs

@@ -1,10 +1,13 @@
 use crate::service::AppService;
 use enclave_api::EnclaveProtoAPI;
+use lcp_proto::google::protobuf::Any;
+use lcp_proto::lcp::service::elc::v1::msg_update_client_stream_chunk::Chunk;
 use lcp_proto::lcp::service::elc::v1::{
     msg_server::Msg, query_server::Query, MsgAggregateMessages, MsgAggregateMessagesResponse,
     MsgCreateClient, MsgCreateClientResponse, MsgUpdateClient, MsgUpdateClientResponse,
-    MsgVerifyMembership, MsgVerifyMembershipResponse, MsgVerifyNonMembership,
-    MsgVerifyNonMembershipResponse, QueryClientRequest, QueryClientResponse,
+    MsgUpdateClientStreamChunk, MsgVerifyMembership, MsgVerifyMembershipResponse,
+    MsgVerifyNonMembership, MsgVerifyNonMembershipResponse, QueryClientRequest,
+    QueryClientResponse,
 };
 use store::transaction::CommitStore;
 use tonic::{Request, Response, Status, Streaming};
@@ -35,30 +38,63 @@ where
         }
     }
 
-    async fn streaming_update_client(
+    async fn update_client_stream(
         &self,
-        request: Request<Streaming<MsgUpdateClient>>,
+        request: Request<Streaming<MsgUpdateClientStreamChunk>>,
     ) -> Result<Response<MsgUpdateClientResponse>, Status> {
-        let mut complete = MsgUpdateClient {
-            signer: vec![],
-            client_id: "".to_string(),
-            include_state: false,
-            header: None,
+        let mut stream = request.into_inner();
+        // read the first message (must be Init)
+        let init = match stream.message().await? {
+            Some(chunk) => match chunk.chunk {
+                Some(Chunk::Init(init)) => init,
+                _ => {
+                    return Err(Status::invalid_argument(
+                        "first message must be of type Init",
+                    ))
+                }
+            },
+            None => {
+                return Err(Status::invalid_argument(
+                    "expected Init message as the first message",
+                ))
+            }
         };
 
-        let mut stream = request.into_inner();
-        while let Some(chunk) = stream.message().await? {
-            if let Some(header) = &mut complete.header {
-                let any_header = chunk
-                    .header
-                    .ok_or(Status::invalid_argument("header value is required"))?;
-                header.value.extend(any_header.value);
-            } else {
-                complete = chunk;
+        // accumulate header chunks
+        let mut header_bytes = Vec::new();
+
+        while let Some(chunk_msg) = stream.message().await? {
+            match chunk_msg.chunk {
+                Some(Chunk::HeaderChunk(header_chunk)) => {
+                    header_bytes.extend(header_chunk.data);
+                }
+                Some(Chunk::Init(_)) => {
+                    return Err(Status::invalid_argument(
+                        "Init must only appear as the first message",
+                    ));
+                }
+                None => {
+                    return Err(Status::invalid_argument("received empty chunk message"));
+                }
             }
         }
 
-        match self.enclave.proto_update_client(complete) {
+        if header_bytes.is_empty() {
+            return Err(Status::invalid_argument("no header data received"));
+        }
+
+        // set the header with collected data
+        let msg = MsgUpdateClient {
+            client_id: init.client_id,
+            include_state: init.include_state,
+            signer: init.signer,
+            header: Some(Any {
+                type_url: init.type_url,
+                value: header_bytes,
+            }),
+        };
+
+        match self.enclave.proto_update_client(msg) {
             Ok(res) => Ok(Response::new(res)),
             Err(e) => Err(Status::aborted(e.to_string())),
         }

proto/definitions/lcp/service/elc/v1/tx.proto

@@ -16,8 +16,8 @@ service Msg {
   // UpdateClient defines a rpc handler method for MsgUpdateClient.
   rpc UpdateClient(MsgUpdateClient) returns (MsgUpdateClientResponse);
 
-  // StreamingUpdateClient defines a rpc handler method for MsgUpdateClient.
-  rpc StreamingUpdateClient(stream MsgUpdateClient) returns (MsgUpdateClientResponse);
+  // UpdateClientStream defines a rpc handler method for MsgUpdateClient.
+  rpc UpdateClientStream(stream MsgUpdateClientStreamChunk) returns (MsgUpdateClientResponse);
 
   // AggregateMessages defines a rpc handler method for MsgAggregateMessages
   rpc AggregateMessages(MsgAggregateMessages) returns (MsgAggregateMessagesResponse);
@@ -137,3 +137,21 @@ message MsgVerifyNonMembershipResponse {
   bytes message = 1;
   bytes signature = 2;
 }
+
+message MsgUpdateClientStreamChunk {
+  oneof chunk {
+    UpdateClientStreamInit init = 1;
+    UpdateClientStreamHeaderChunk header_chunk = 2;
+  }
+}
+
+message UpdateClientStreamInit {
+  string client_id = 1;
+  bool include_state = 2;
+  bytes signer = 3;
+  string type_url = 4;
+}
+
+message UpdateClientStreamHeaderChunk {
+  bytes data = 1;
+}

proto/src/descriptor.bin

@@ -11,10 +11,8 @@ pub struct Plan {
     /// reached and the software will exit.
     #[prost(string, tag = "1")]
     pub name: ::prost::alloc::string::String,
-    /// Deprecated: Time based upgrades have been deprecated. Time based upgrade logic
-    /// has been removed from the SDK.
-    /// If this field is not empty, an error will be thrown.
-    #[deprecated]
+    /// The time after which the upgrade must be performed.
+    /// Leave set to its zero value to use a pre-defined Height instead.
     #[prost(message, optional, tag = "2")]
     pub time: ::core::option::Option<super::super::super::google::protobuf::Timestamp>,
     /// The height at which the upgrade must be performed.
@@ -25,14 +23,6 @@ pub struct Plan {
     /// such as a git commit that validators could automatically upgrade to
     #[prost(string, tag = "4")]
     pub info: ::prost::alloc::string::String,
-    /// Deprecated: UpgradedClientState field has been deprecated. IBC upgrade logic has been
-    /// moved to the IBC module in the sub module 02-client.
-    /// If this field is not empty, an error will be thrown.
-    #[deprecated]
-    #[prost(message, optional, tag = "5")]
-    pub upgraded_client_state: ::core::option::Option<
-        super::super::super::google::protobuf::Any,
-    >,
 }
 /// SoftwareUpgradeProposal is a gov Content type for initiating a software
 /// upgrade.
@@ -56,16 +46,3 @@ pub struct CancelSoftwareUpgradeProposal {
     #[prost(string, tag = "2")]
     pub description: ::prost::alloc::string::String,
 }
-/// ModuleVersion specifies a module and its consensus version.
-///
-/// Since: cosmos-sdk 0.43
-#[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct ModuleVersion {
-    /// name of the app module
-    #[prost(string, tag = "1")]
-    pub name: ::prost::alloc::string::String,
-    /// consensus version of the app module
-    #[prost(uint64, tag = "2")]
-    pub version: u64,
-}