RPC support for CometBFT 0.38

.changelog/unreleased/breaking-changes/1317-cometbft-rpc-0.38.md

@@ -0,0 +1,23 @@
+- `[tendermint-rpc]`  Changes to support the RPC protocol in CometBFT 0.38
+  ([\#1317](https://github.com/informalsystems/tendermint-rs/pull/1317)):
+  * Add `finalize_block_results` and `app_hash` fields to
+    `endpoint::block_results::Response`.
+  * The `deliver_tx` field is renamed to `tx_result` in
+    `endpoint::broadcast::tx_commit::Response`.
+  * The `tx_result` field type changed to `ExecTxResult` in
+    `endpoint::tx::Response`.
+  * The `event::EventData::NewBlock` variant is renamed to `LegacyNewBlock`.
+    The new `NewBlock` variant only carries fields relevant since CometBFT 0.38.
+  * Removed `event::DialectEvent`, replaced with `event::v0_34::DialectEvent`
+    and `event::latest::DialectEvent` as non-generic serialization helpers.
+    The latter handles the fields added in CometBFT 0.38, `block_id` and
+    `result_finalize_block`. Same refactoring done for `DialectEventData`
+    and other types used in the event data structures.
+  * Changed some of the serialization dialect helpers only be
+    used by the 0.34 dialect and remove generics. The current dialect's
+    seralization is switched to the serde impls on the domain types in
+    `tendermint`.
+- `[tendermint]` Changes to support the RPC protocol in CometBFT 0.38
+  ([\#1317](https://github.com/informalsystems/tendermint-rs/pull/1317)):
+  * Due to some attribute changes, the format emitted by `Serialize` is
+    changed for `abci::response` types `CheckTx` and `FinalizeBlock`.

.changelog/unreleased/improvements/1317-cometbft-rpc-0.38.md

@@ -0,0 +1,9 @@
+- `[tendermint-rpc]` Support for CometBFT 0.38
+  ([\#1317](https://github.com/informalsystems/tendermint-rs/pull/1317)):
+  * `Deserialize` implementations on `abci::Event`, `abci::EventAttribute`
+    that correspond to the current RPC serialization.
+  * Domain types under `abci::response` also get `Deserialize` implementations
+    corresponding to the current RPC serialization.
+  * `Serialize`, `Deserialize` implementations on `abci::types::ExecTxResult`
+    corresponding to the current RPC serialization.
+  * Added the `apphash_base64` serializer module.

rpc/src/client/bin/main.rs

@@ -6,12 +6,8 @@ use futures::StreamExt;
 use structopt::StructOpt;
 use tendermint::Hash;
 use tendermint_rpc::{
-    client::CompatMode,
-    dialect::{Dialect, LatestDialect},
-    event::DialectEvent,
-    query::Query,
-    Client, Error, HttpClient, Order, Paging, Scheme, Subscription, SubscriptionClient, Url,
-    WebSocketClient,
+    client::CompatMode, event::latest::DialectEvent, query::Query, Client, Error, HttpClient,
+    Order, Paging, Scheme, Subscription, SubscriptionClient, Url, WebSocketClient,
 };
 use tokio::{task::JoinHandle, time::Duration};
 use tracing::{debug, error, info, level_filters::LevelFilter, warn};
@@ -504,7 +500,7 @@ async fn recv_events_with_timeout(
                     }
                 };
                 let event = result?;
-                let event: DialectEvent<<LatestDialect as Dialect>::Event> = event.into();
+                let event: DialectEvent = event.into();
                 println!("{}", serde_json::to_string_pretty(&event).map_err(Error::serde)?);
                 event_count += 1;
                 if let Some(me) = max_events {
@@ -526,7 +522,7 @@ async fn recv_events(mut subs: Subscription, max_events: Option<u32>) -> Result<
     let mut event_count = 0u64;
     while let Some(result) = subs.next().await {
         let event = result?;
-        let event: DialectEvent<<LatestDialect as Dialect>::Event> = event.into();
+        let event: DialectEvent = event.into();
         println!(
             "{}",
             serde_json::to_string_pretty(&event).map_err(Error::serde)?

rpc/src/client/compat.rs

@@ -49,6 +49,7 @@ impl CompatMode {
         match (version.major, version.minor) {
             (0, 34) => Ok(CompatMode::V0_34),
             (0, 37) => Ok(CompatMode::V0_37),
+            (0, 38) => Ok(CompatMode::V0_37),
             _ => Err(Error::unsupported_tendermint_version(version.to_string())),
         }
     }
@@ -88,7 +89,11 @@ mod tests {
             CompatMode::from_version(parse_version("v0.37.0")).unwrap(),
             CompatMode::V0_37
         );
-        let res = CompatMode::from_version(parse_version("v0.38.0"));
+        assert_eq!(
+            CompatMode::from_version(parse_version("v0.38.0")).unwrap(),
+            CompatMode::V0_37
+        );
+        let res = CompatMode::from_version(parse_version("v0.39.0"));
         assert!(res.is_err());
         let res = CompatMode::from_version(parse_version("v1.0.0"));
         assert!(res.is_err());

rpc/src/client/transport/mock.rs

@@ -264,12 +264,10 @@ mod test {
 
     mod v0_34 {
         use super::*;
-        use crate::dialect::v0_34::Event as RpcEvent;
-        use crate::event::DialectEvent;
+        use crate::event::v0_34::DialectEvent;
 
         async fn read_event(name: &str) -> Event {
-            let msg = DialectEvent::<RpcEvent>::from_string(read_json_fixture("v0_34", name).await)
-                .unwrap();
+            let msg = DialectEvent::from_string(read_json_fixture("v0_34", name).await).unwrap();
             msg.into()
         }
 
@@ -334,12 +332,10 @@ mod test {
 
     mod v0_37 {
         use super::*;
-        use crate::dialect::v0_37::Event as RpcEvent;
-        use crate::event::DialectEvent;
+        use crate::event::latest::DialectEvent;
 
         async fn read_event(name: &str) -> Event {
-            let msg = DialectEvent::<RpcEvent>::from_string(read_json_fixture("v0_37", name).await)
-                .unwrap();
+            let msg = DialectEvent::from_string(read_json_fixture("v0_37", name).await).unwrap();
             msg.into()
         }
 

rpc/src/client/transport/router.rs

@@ -144,7 +144,7 @@ mod test {
     use super::*;
     use crate::{
         client::sync::{unbounded, ChannelRx},
-        event::{Event, WrappedEvent},
+        event::Event,
         utils::uuid_str,
     };
 
@@ -180,16 +180,15 @@ mod test {
 
     mod v0_34 {
         use super::*;
-        use crate::dialect::v0_34::Event as RpcEvent;
+
+        type WrappedEvent = crate::response::Wrapper<crate::event::v0_34::DialectEvent>;
 
         async fn read_event(name: &str) -> Event {
-            serde_json::from_str::<WrappedEvent<RpcEvent>>(
-                read_json_fixture("v0_34", name).await.as_str(),
-            )
-            .unwrap()
-            .into_result()
-            .unwrap()
-            .into()
+            serde_json::from_str::<WrappedEvent>(read_json_fixture("v0_34", name).await.as_str())
+                .unwrap()
+                .into_result()
+                .unwrap()
+                .into()
         }
 
         #[tokio::test]
@@ -229,16 +228,15 @@ mod test {
 
     mod v0_37 {
         use super::*;
-        use crate::dialect::v0_37::Event as RpcEvent;
+
+        type WrappedEvent = crate::response::Wrapper<crate::event::latest::DialectEvent>;
 
         async fn read_event(name: &str) -> Event {
-            serde_json::from_str::<WrappedEvent<RpcEvent>>(
-                read_json_fixture("v0_37", name).await.as_str(),
-            )
-            .unwrap()
-            .into_result()
-            .unwrap()
-            .into()
+            serde_json::from_str::<WrappedEvent>(read_json_fixture("v0_37", name).await.as_str())
+                .unwrap()
+                .into_result()
+                .unwrap()
+                .into()
         }
 
         #[tokio::test]

rpc/src/client/transport/websocket.rs

@@ -25,7 +25,7 @@ use tendermint::{block::Height, Hash};
 use tendermint_config::net;
 
 use super::router::{SubscriptionId, SubscriptionIdRef};
-use crate::dialect::{v0_34, v0_37};
+use crate::dialect::v0_34;
 use crate::{
     client::{
         subscription::SubscriptionTx,
@@ -35,7 +35,7 @@ use crate::{
     },
     endpoint::{self, subscribe, unsubscribe},
     error::Error,
-    event::{DialectEvent, Event},
+    event::{self, Event},
     prelude::*,
     query::Query,
     request::Wrapper,
@@ -883,8 +883,8 @@ impl WebSocketClientDriver {
 
     async fn handle_text_msg(&mut self, msg: String) -> Result<(), Error> {
         let parse_res = match self.compat {
-            CompatMode::V0_37 => DialectEvent::<v0_37::Event>::from_string(&msg).map(Into::into),
-            CompatMode::V0_34 => DialectEvent::<v0_34::Event>::from_string(&msg).map(Into::into),
+            CompatMode::V0_37 => event::v0_37::DialectEvent::from_string(&msg).map(Into::into),
+            CompatMode::V0_34 => event::v0_34::DialectEvent::from_string(&msg).map(Into::into),
         };
         if let Ok(ev) = parse_res {
             debug!("JSON-RPC event: {}", msg);
@@ -1024,7 +1024,7 @@ mod test {
     };
 
     use super::*;
-    use crate::{client::sync::unbounded, dialect, query::EventType, request, Id, Method};
+    use crate::{client::sync::unbounded, event, query::EventType, request, Id, Method};
 
     // Interface to a driver that manages all incoming WebSocket connections.
     struct TestServer {
@@ -1214,11 +1214,11 @@ mod test {
             };
             match self.compat {
                 CompatMode::V0_37 => {
-                    let ev: DialectEvent<dialect::v0_37::Event> = ev.into();
+                    let ev: event::v0_37::DialectEvent = ev.into();
                     self.send(subs_id, ev).await;
                 },
                 CompatMode::V0_34 => {
-                    let ev: DialectEvent<dialect::v0_34::Event> = ev.into();
+                    let ev: event::v0_34::DialectEvent = ev.into();
                     self.send(subs_id, ev).await;
                 },
             }
@@ -1333,10 +1333,10 @@ mod test {
 
     mod v0_34 {
         use super::*;
-        use crate::dialect::v0_34::Event as RpcEvent;
+        use crate::event::v0_34::DialectEvent;
 
         async fn read_event(name: &str) -> Event {
-            DialectEvent::<RpcEvent>::from_string(read_json_fixture("v0_34", name).await)
+            DialectEvent::from_string(read_json_fixture("v0_34", name).await)
                 .unwrap()
                 .into()
         }
@@ -1400,10 +1400,77 @@ mod test {
 
     mod v0_37 {
         use super::*;
-        use crate::dialect::v0_37::Event as RpcEvent;
+        use crate::event::latest::DialectEvent;
 
         async fn read_event(name: &str) -> Event {
-            DialectEvent::<RpcEvent>::from_string(read_json_fixture("v0_37", name).await)
+            DialectEvent::from_string(read_json_fixture("v0_37", name).await)
+                .unwrap()
+                .into()
+        }
+
+        #[tokio::test]
+        async fn websocket_client_happy_path() {
+            let event1 = read_event("subscribe_newblock_0").await;
+            let event2 = read_event("subscribe_newblock_1").await;
+            let event3 = read_event("subscribe_newblock_2").await;
+            let test_events = vec![event1, event2, event3];
+
+            println!("Starting WebSocket server...");
+            let mut server = TestServer::new("127.0.0.1:0", CompatMode::V0_37).await;
+            println!("Creating client RPC WebSocket connection...");
+            let url = server.node_addr.clone().try_into().unwrap();
+            let (client, driver) = WebSocketClient::builder(url)
+                .compat_mode(CompatMode::V0_37)
+                .build()
+                .await
+                .unwrap();
+            let driver_handle = tokio::spawn(async move { driver.run().await });
+
+            println!("Initiating subscription for new blocks...");
+            let mut subs = client.subscribe(EventType::NewBlock.into()).await.unwrap();
+
+            // Collect all the events from the subscription.
+            let subs_collector_hdl = tokio::spawn(async move {
+                let mut results = Vec::new();
+                while let Some(res) = subs.next().await {
+                    results.push(res);
+                    if results.len() == 3 {
+                        break;
+                    }
+                }
+                results
+            });
+
+            println!("Publishing events");
+            // Publish the events from this context
+            for ev in &test_events {
+                server.publish_event(ev.clone()).unwrap();
+            }
+
+            println!("Collecting results from subscription...");
+            let collected_results = subs_collector_hdl.await.unwrap();
+
+            client.close().unwrap();
+            server.terminate().await.unwrap();
+            let _ = driver_handle.await.unwrap();
+            println!("Closed client and terminated server");
+
+            assert_eq!(3, collected_results.len());
+            for i in 0..3 {
+                assert_eq!(
+                    test_events[i],
+                    collected_results[i].as_ref().unwrap().clone()
+                );
+            }
+        }
+    }
+
+    mod v0_38 {
+        use super::*;
+        use crate::event::latest::DialectEvent;
+
+        async fn read_event(name: &str) -> Event {
+            DialectEvent::from_string(read_json_fixture("v0_38", name).await)
                 .unwrap()
                 .into()
         }

rpc/src/dialect/deliver_tx.rs

@@ -70,3 +70,21 @@ where
         }
     }
 }
+
+impl<Ev> From<DeliverTx<Ev>> for abci::types::ExecTxResult
+where
+    Ev: Into<abci::Event>,
+{
+    fn from(msg: DeliverTx<Ev>) -> Self {
+        Self {
+            code: msg.code,
+            data: msg.data,
+            log: msg.log,
+            info: msg.info,
+            gas_wanted: msg.gas_wanted,
+            gas_used: msg.gas_used,
+            events: msg.events.into_iter().map(Into::into).collect(),
+            codespace: msg.codespace,
+        }
+    }
+}