rpc: Add /genesis_chunked query (#1439)

* Add /genesis_chunked query

* Add changelog entry

* Fix comments

* Update types of GenesisChunked request and response

* Convert chunk and total fields from GenesisChunked response

* Improve genesis_chunked_stream method

* Add genesis-chunked CLI

* Fix dependencies for genesis_chunked_stream

* Update changelog entry

---------

Co-authored-by: Romain Ruetschi <romain@informal.systems>

.changelog/unreleased/features/1438-genesis-chunked-endpoint.md

@@ -0,0 +1,2 @@
+- `[tendermint-rpc]` Add support for the `/genesis_chunked` RPC endpoint
+  ([\#1438](https://github.com/informalsystems/tendermint-rs/issues/1438))

rpc/src/client.rs

@@ -271,6 +271,34 @@ pub trait Client {
         Ok(self.perform(genesis::Request::default()).await?.genesis)
     }
 
+    async fn genesis_chunked(&self, chunk: u64) -> Result<genesis_chunked::Response, Error> {
+        self.perform(genesis_chunked::Request::new(chunk)).await
+    }
+
+    /// `/genesis_chunked`: get genesis file in multiple chunks.
+    #[cfg(any(feature = "http-client", feature = "websocket-client"))]
+    async fn genesis_chunked_stream(
+        &self,
+    ) -> core::pin::Pin<Box<dyn futures::Stream<Item = Result<Vec<u8>, Error>> + '_>> {
+        Box::pin(futures::stream::unfold(Some(0), move |chunk| async move {
+            // Verify if there are more chunks to fetch
+            let chunk = chunk?;
+
+            match self.genesis_chunked(chunk).await {
+                Ok(response) => {
+                    if response.chunk + 1 >= response.total {
+                        // No more chunks to fetch
+                        Some((Ok(response.data), None))
+                    } else {
+                        // Emit this chunk and fetch the next chunk
+                        Some((Ok(response.data), Some(response.chunk + 1)))
+                    }
+                },
+                Err(e) => Some((Err(e), None)), // Abort the stream
+            }
+        }))
+    }
+
     /// `/net_info`: obtain information about P2P and other network connections.
     async fn net_info(&self) -> Result<net_info::Response, Error> {
         self.perform(net_info::Request).await

rpc/src/client/bin/main.rs

@@ -136,6 +136,8 @@ enum ClientRequest {
     ConsensusState,
     /// Get the node's genesis data.
     Genesis,
+    /// Get the node's genesis data by chunks
+    GenesisChunked,
     /// Get the node's health.
     Health,
     /// Request the latest block.
@@ -413,6 +415,20 @@ where
             serde_json::to_string_pretty(&client.genesis::<serde_json::Value>().await?)
                 .map_err(Error::serde)?
         },
+        ClientRequest::GenesisChunked => {
+            let mut data = Vec::new();
+            let mut chunks = client.genesis_chunked_stream().await;
+
+            while let Some(chunk) = chunks.next().await {
+                let mut chunk = chunk?;
+                data.append(&mut chunk);
+            }
+
+            let genesis: tendermint::genesis::Genesis<serde_json::Value> =
+                serde_json::from_slice(&data).map_err(Error::serde)?;
+
+            serde_json::to_string_pretty(&genesis).map_err(Error::serde)?
+        },
         ClientRequest::Health => {
             serde_json::to_string_pretty(&client.health().await?).map_err(Error::serde)?
         },

rpc/src/endpoint.rs

@@ -13,6 +13,7 @@ pub mod consensus_params;
 pub mod consensus_state;
 pub mod evidence;
 pub mod genesis;
+pub mod genesis_chunked;
 pub mod header;
 pub mod header_by_hash;
 pub mod health;

rpc/src/endpoint/genesis_chunked.rs

@@ -0,0 +1,57 @@
+//! `/genesis_chunked` endpoint JSON-RPC wrapper
+
+use alloc::{
+    string::{String, ToString},
+    vec::Vec,
+};
+use serde::{Deserialize, Serialize};
+use tendermint_proto::serializers;
+
+use crate::{dialect::Dialect, request::RequestMessage};
+
+/// Get the genesis state in multiple chunks for the current chain
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct Request {
+    pub chunk: String,
+}
+
+impl Request {
+    pub fn new(chunk: u64) -> Self {
+        Self {
+            chunk: chunk.to_string(),
+        }
+    }
+}
+
+impl RequestMessage for Request {
+    fn method(&self) -> crate::Method {
+        crate::Method::GenesisChunked
+    }
+}
+
+impl<S> crate::Request<S> for Request
+where
+    S: Dialect,
+{
+    type Response = Response;
+}
+
+impl<S> crate::SimpleRequest<S> for Request
+where
+    S: Dialect,
+{
+    type Output = Response;
+}
+
+/// Block responses
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Response {
+    #[serde(with = "serializers::from_str")]
+    pub chunk: u64,
+    #[serde(with = "serializers::from_str")]
+    pub total: u64,
+    #[serde(with = "serializers::bytes::base64string")]
+    pub data: Vec<u8>,
+}
+
+impl crate::Response for Response {}

rpc/src/method.rs

@@ -56,6 +56,9 @@ pub enum Method {
     /// Get genesis file
     Genesis,
 
+    /// Get genesis file in multiple chunks
+    GenesisChunked,
+
     /// Get block header
     Header,
 
@@ -109,6 +112,7 @@ impl Method {
             Method::ConsensusParams => "consensus_params",
             Method::ConsensusState => "consensus_state",
             Method::Genesis => "genesis",
+            Method::GenesisChunked => "genesis_chunked",
             Method::Header => "header",
             Method::HeaderByHash => "header_by_hash",
             Method::Health => "health",