feat(proto)!: grpc server and client support

abci/Cargo.toml

@@ -20,6 +20,8 @@ server = [
     "dep:tokio-util",
     "dep:futures",
 ]
+
+grpc = ["tenderdash-proto/grpc"]
 crypto = ["dep:lhash"]
 tcp = ["server"]
 unix = ["server"]
@@ -66,3 +68,5 @@ futures = { version = "0.3.26" }
 tokio = { version = "1", features = ["macros", "signal", "time", "io-std"] }
 hex = { version = "0.4" }
 lazy_static = { version = "1.4" }
+# For tests of gRPC server
+tonic = { version = "0.11" }

abci/src/lib.rs

@@ -22,7 +22,7 @@ pub use application::{check_version, Application, RequestDispatcher};
 use prost::{DecodeError, EncodeError};
 #[allow(deprecated)]
 #[cfg(feature = "server")]
-pub use server::{start_server, CancellationToken, Server, ServerBuilder};
+pub use server::{start_server, CancellationToken, Server, ServerBuilder, ServerRuntime};
 pub use tenderdash_proto as proto;
 
 #[cfg(feature = "crypto")]

abci/src/server.rs

@@ -8,11 +8,15 @@ use std::{
     str::FromStr,
 };
 
+use futures::Future;
 #[cfg(feature = "tcp")]
 use tokio::net::TcpListener;
 #[cfg(feature = "unix")]
 use tokio::net::UnixListener;
-use tokio::runtime::{Handle, Runtime};
+use tokio::{
+    runtime::{Handle, Runtime},
+    task::JoinHandle,
+};
 pub use tokio_util::sync::CancellationToken;
 
 use self::generic::GenericServer;
@@ -208,10 +212,24 @@ impl<'a, App: RequestDispatcher + 'a> ServerBuilder<App> {
 }
 
 /// Server runtime that must be alive for the whole lifespan of the server
-pub(crate) struct ServerRuntime {
+pub struct ServerRuntime {
     /// Runtime stored here to ensure it is never dropped
     _runtime: Option<Runtime>,
-    handle: Handle,
+    pub handle: Handle,
+}
+
+impl ServerRuntime {
+    pub fn block_on<F: std::future::Future>(&self, future: F) -> F::Output {
+        self.handle.block_on(future)
+    }
+
+    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        self.handle.spawn(future)
+    }
 }
 
 impl Default for ServerRuntime {

abci/tests/common/docker.rs

@@ -6,7 +6,8 @@ use bollard::{
     Docker, API_DEFAULT_VERSION,
 };
 use futures::StreamExt;
-use tokio::{io::AsyncWriteExt, runtime::Runtime, time::timeout};
+use tenderdash_abci::ServerRuntime;
+use tokio::{io::AsyncWriteExt, time::timeout};
 use tracing::{debug, error, info};
 use url::Url;
 
@@ -16,7 +17,7 @@ pub struct TenderdashDocker {
     name: String,
     docker: Docker,
     image: String,
-    runtime: Runtime,
+    runtime: ServerRuntime,
 }
 impl TenderdashDocker {
     /// new() creates and starts new Tenderdash docker container for provided
@@ -31,8 +32,8 @@ impl TenderdashDocker {
     ///
     /// * `tag` - Docker tag to use; provide empty string to use default
     /// * `app_address` - address of ABCI app server; for example,
-    ///   `tcp://172.17.0.1:4567`, `tcp://[::ffff:ac11:1]:5678` or
-    ///   `unix:///path/to/file`
+    ///   `tcp://172.17.0.1:4567`, `tcp://[::ffff:ac11:1]:5678`,
+    ///   `grpc://172.17.01:5678` or `unix:///path/to/file`
     pub(crate) fn new(
         container_name: &str,
         tag: Option<&str>,
@@ -45,14 +46,14 @@ impl TenderdashDocker {
         };
 
         let app_address = url::Url::parse(app_address).expect("invalid app address");
-        if app_address.scheme() != "tcp" && app_address.scheme() != "unix" {
-            panic!("app_address must be either tcp:// or unix://");
+        if app_address.scheme() != "tcp"
+            && app_address.scheme() != "unix"
+            && app_address.scheme() != "grpc"
+        {
+            panic!("app_address must be either grpc://, tcp:// or unix://");
         }
 
-        let runtime = tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
-            .build()
-            .expect("cannot initialize tokio runtime");
+        let runtime = tenderdash_abci::ServerRuntime::default();
 
         info!("Starting Tenderdash docker container");
 
@@ -152,12 +153,24 @@ impl TenderdashDocker {
             None
         };
 
-        let app_address = app_address.to_string().replace('/', "\\/");
+        let (abci, app_address) = match app_address.scheme() {
+            "grpc" => {
+                let address = app_address
+                    .to_string()
+                    .replace("grpc://", "")
+                    .replace('/', "\\/");
+                ("grpc", address)
+            },
+            _ => ("socket", app_address.to_string().replace('/', "\\/")),
+        };
 
         debug!("Tenderdash will connect to ABCI address: {}", app_address);
         let container_config = Config {
             image: Some(self.image.clone()),
-            env: Some(vec![format!("PROXY_APP={}", app_address)]),
+            env: Some(vec![
+                format!("PROXY_APP={}", app_address),
+                format!("ABCI={}", abci),
+            ]),
             host_config: Some(HostConfig {
                 binds,
                 ..Default::default()
@@ -263,6 +276,7 @@ impl Drop for TenderdashDocker {
         }
     }
 }
+
 /// Use custom panic handler to dump logs on panic
 #[allow(dead_code)]
 pub fn setup_td_logs_panic(td_docker: &Arc<TenderdashDocker>) {

abci/tests/grpc.rs

@@ -0,0 +1,150 @@
+//! Test gRPC server for ABCI protocol.
+//!
+//! This test verifies that the gRPC server generated with tonic as part of the
+//! tenderdash-proto crate can successfully connect to Tenderdash instance.
+//!
+//! This test should be implemented in the tenderdash-proto crate; however, it
+//! is implemented here to use already existing docker container testing
+//! logic.
+#![cfg(feature = "grpc")]
+
+use std::sync::Arc;
+
+use tenderdash_abci::{
+    proto::abci::{
+        abci_application_server::AbciApplication, RequestEcho, RequestInfo, ResponseInfo,
+    },
+    CancellationToken,
+};
+mod common;
+use tenderdash_abci::proto;
+use tonic::{async_trait, Response, Status};
+
+#[cfg(feature = "docker-tests")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+/// Test server listening on ipv4 address.
+///
+/// See [tcp_server_test()].
+async fn test_ipv4_server() {
+    // we assume the host uses default Docker network configuration, with the host
+    // using 172.17.0.1
+    let bind_address = "172.17.0.1:1234".to_string();
+
+    grpc_server_test("v4", bind_address.as_str()).await;
+}
+
+#[cfg(feature = "docker-tests")]
+#[tokio::test]
+/// Test server listening on ipv6 address.
+///
+/// See [tcp_server_test()].
+async fn test_ipv6_server() {
+    // we assume the host uses default Docker network configuration, with the host
+    // using 172.17.0.1. This is IPv6 notation of the IPv4 address.
+    let bind_address = "[::ffff:ac11:1]:5678".to_string();
+
+    grpc_server_test("v6", bind_address.as_str()).await;
+}
+
+#[cfg(feature = "docker-tests")]
+/// Feature: ABCI App TCO server
+///
+/// * Given that we have Tenderdash instance using TCP connection to communicate
+///   with ABCI APP
+/// * When we estabilish connection with Tenderdash
+/// * Then Tenderdash sends Info request
+async fn grpc_server_test(test_name: &str, bind_address: &str) {
+    use core::panic;
+
+    use proto::abci::abci_application_server::AbciApplicationServer;
+    use tonic::transport::Server;
+
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::new("debug"))
+        .with_ansi(true)
+        .try_init()
+        .ok();
+
+    let cancel = CancellationToken::new();
+    let app = TestApp {
+        cancel: cancel.clone(),
+    };
+
+    let addr = bind_address.parse().expect("address must be valid");
+    let server_cancel = cancel.clone();
+    let server_handle = tokio::spawn(async move {
+        tracing::debug!("starting gRPC server");
+        Server::builder()
+            .add_service(AbciApplicationServer::new(app))
+            .serve_with_shutdown(addr, server_cancel.cancelled())
+            .await
+            .expect("server failed");
+        tracing::debug!("gRPC server stopped");
+    });
+
+    let socket_uri = format!("grpc://{}", bind_address);
+    let container_name = format!("tenderdash_{}", test_name);
+
+    let td = tokio::task::spawn_blocking(move || {
+        tracing::debug!("starting Tenderdash in Docker container");
+        let td = Arc::new(common::docker::TenderdashDocker::new(
+            &container_name,
+            Some("feat-ABCI-protocol-env-var"),
+            &socket_uri,
+        ));
+        common::docker::setup_td_logs_panic(&td);
+        tracing::debug!("started Tenderdash in Docker container");
+
+        td
+    })
+    .await
+    .expect("start tenderdash");
+
+    // tracing::debug!(?result, "connection handled");
+    // assert!(matches!(result, Ok(())));
+    tokio::select! {
+        _ = tokio::time::sleep(tokio::time::Duration::from_secs(60)) => {
+            panic!("Test timed out");
+        }
+        _ = cancel.cancelled() => {
+            tracing::debug!("CancellationToken cancelled");
+        }
+        ret = server_handle => {
+            ret.expect("gRPC server failed");
+        }
+    }
+
+    tokio::task::spawn_blocking(move || drop(td))
+        .await
+        .expect("tenderdash cleanup");
+
+    tracing::info!("Test finished successfully");
+}
+
+pub struct TestApp {
+    // when test succeeds, we cancel this token to finish it
+    cancel: CancellationToken,
+}
+#[async_trait]
+impl AbciApplication for TestApp {
+    async fn echo(
+        &self,
+        request: tonic::Request<RequestEcho>,
+    ) -> Result<tonic::Response<proto::abci::ResponseEcho>, Status> {
+        tracing::info!(?request, "Echo called");
+        Ok(Response::new(proto::abci::ResponseEcho {
+            message: request.into_inner().message,
+        }))
+    }
+    async fn info(
+        &self,
+        _request: tonic::Request<RequestInfo>,
+    ) -> std::result::Result<tonic::Response<ResponseInfo>, tonic::Status> {
+        tracing::info!("Info called, test successful");
+        let resp = ResponseInfo {
+            ..Default::default()
+        };
+        self.cancel.cancel();
+        Ok(Response::new(resp))
+    }
+}

proto-compiler/Cargo.toml

@@ -17,3 +17,15 @@ regex = { "version" = "1.7.1" }
 ureq = { "version" = "2.6.2" }
 zip = { version = "0.6.4", default-features = false, features = ["deflate"] }
 fs_extra = { version = "1.3.0" }
+tonic-build = { version = "0.11.0", optional = true }
+
+
+[features]
+default = []
+# Enable gRPC support; needed by server and client features.
+# Conflicts with no_std
+grpc = ["dep:tonic-build"]
+# Build the gRPC server. Requires tenderdash-proto/std feature.
+server = ["grpc"]
+# Build the gRPC client. Requires tenderdash-proto/std feature.
+client = ["grpc"]

proto-compiler/src/constants.rs

@@ -54,7 +54,6 @@ const DERIVE_FROM_STR: &str = r#"#[derive(derive_more::FromStr)]"#;
 /// here: <https://docs.rs/prost-build/0.6.1/prost_build/struct.Config.html#method.btree_map>
 pub static CUSTOM_TYPE_ATTRIBUTES: &[(&str, &str)] = &[
     (".tendermint.libs.bits.BitArray", SERIALIZED),
-    (".tendermint.types.EvidenceParams", SERIALIZED),
     (".tendermint.types.BlockIDFlag", PRIMITIVE_ENUM),
     (".tendermint.types.Block", SERIALIZED),
     (".tendermint.types.Data", SERIALIZED),
@@ -91,6 +90,15 @@ pub static CUSTOM_TYPE_ATTRIBUTES: &[(&str, &str)] = &[
     (".tendermint.crypto.Proof", SERIALIZED),
     (".tendermint.abci.Response.value", DERIVE_FROM),
     (".tendermint.abci.Request.value", DERIVE_FROM),
+    // Consensus params
+    (".tendermint.types.ConsensusParams", SERIALIZED),
+    (".tendermint.types.ABCIParams", SERIALIZED),
+    (".tendermint.types.BlockParams", SERIALIZED),
+    (".tendermint.types.EvidenceParams", SERIALIZED),
+    (".tendermint.types.ValidatorParams", SERIALIZED),
+    (".tendermint.types.VersionParams", SERIALIZED),
+    (".tendermint.types.SynchronyParams", SERIALIZED),
+    (".tendermint.types.TimeoutParams", SERIALIZED),
 ];
 
 /// Custom field attributes applied on top of protobuf fields in (a) struct(s)
@@ -99,10 +107,6 @@ pub static CUSTOM_TYPE_ATTRIBUTES: &[(&str, &str)] = &[
 /// The first item is a path as defined in the prost_build::Config::btree_map
 /// here: <https://docs.rs/prost-build/0.6.1/prost_build/struct.Config.html#method.btree_map>
 pub static CUSTOM_FIELD_ATTRIBUTES: &[(&str, &str)] = &[
-    (
-        ".tendermint.types.EvidenceParams.max_bytes",
-        QUOTED_WITH_DEFAULT,
-    ),
     (".tendermint.version.Consensus.block", QUOTED),
     (".tendermint.version.Consensus.app", QUOTED_WITH_DEFAULT),
     (".tendermint.abci.ResponseInfo.data", DEFAULT),
@@ -200,4 +204,22 @@ pub static CUSTOM_FIELD_ATTRIBUTES: &[(&str, &str)] = &[
     (".tendermint.crypto.Proof.total", QUOTED),
     (".tendermint.crypto.Proof.aunts", VEC_BASE64STRING),
     (".tendermint.crypto.Proof.leaf_hash", BASE64STRING),
+    // Consensus params
+    (
+        ".tendermint.types.BlockParams.max_bytes",
+        QUOTED_WITH_DEFAULT,
+    ),
+    (".tendermint.types.BlockParams.max_gas", QUOTED_WITH_DEFAULT),
+    (
+        ".tendermint.types.EvidenceParams.max_age_num_blocks",
+        QUOTED_WITH_DEFAULT,
+    ),
+    (
+        ".tendermint.types.EvidenceParams.max_bytes",
+        QUOTED_WITH_DEFAULT,
+    ),
+    (
+        ".tendermint.types.VersionParams.app_version",
+        QUOTED_WITH_DEFAULT,
+    ),
 ];

proto-compiler/src/lib.rs

@@ -107,6 +107,14 @@ pub fn proto_compile() {
     let tenderdash_ver = tenderdash_version(tenderdash_dir);
 
     println!("[info] => Creating structs.");
+
+    #[cfg(feature = "grpc")]
+    tonic_build::configure()
+        .generate_default_stubs(true)
+        .compile_with_config(pb, &protos, &proto_includes_paths)
+        .unwrap();
+
+    #[cfg(not(feature = "grpc"))]
     pb.compile_protos(&protos, &proto_includes_paths).unwrap();
 
     println!("[info] => Removing old structs and copying new structs.");