[feature] #3355: Standardize block API (#3884)

Co-authored-by: Shanin Roman <shanin1000@yandex.ru>
[fix] #3890: Fix validate topology on block sync

Signed-off-by: Marin Veršić <marin.versic101@gmail.com>

Author: mversic

Cargo.lock

@@ -14,7 +14,7 @@ use warp::ws::WebSocket;
 use crate::stream::{self, Sink, Stream};
 
 /// Type of Stream error
-pub type StreamError = stream::Error<<WebSocket as Stream<VersionedEventSubscriptionRequest>>::Err>;
+pub type StreamError = stream::Error<<WebSocket as Stream<EventSubscriptionRequest>>::Err>;
 
 /// Type of error for `Consumer`
 #[derive(thiserror::Error, Debug)]
@@ -57,9 +57,7 @@ impl Consumer {
     /// Can fail due to timeout or without message at websocket or during decoding request
     #[iroha_futures::telemetry_future]
     pub async fn new(mut stream: WebSocket) -> Result<Self> {
-        let subscription_request: VersionedEventSubscriptionRequest = stream.recv().await?;
-        let EventSubscriptionRequest(filter) = subscription_request.into_v1();
-
+        let EventSubscriptionRequest(filter) = stream.recv().await?;
         Ok(Consumer { stream, filter })
     }
 
@@ -74,7 +72,7 @@ impl Consumer {
         }
 
         self.stream
-            .send(VersionedEventMessage::from(EventMessage(event)))
+            .send(EventMessage(event))
             .await
             .map_err(Into::into)
     }

cli/src/event.rs

@@ -160,9 +160,9 @@ impl NetworkRelay {
 
         match msg {
             SumeragiPacket(data) => {
-                self.sumeragi.incoming_message(data.into_v1());
+                self.sumeragi.incoming_message(*data);
             }
-            BlockSync(data) => self.block_sync.message(data.into_v1()).await,
+            BlockSync(data) => self.block_sync.message(*data).await,
             TransactionGossiper(data) => self.gossiper.gossip(*data).await,
             Health => {}
         }

cli/src/lib.rs

@@ -6,6 +6,7 @@ use core::{result::Result, time::Duration};
 
 use futures::{SinkExt, StreamExt};
 use iroha_version::prelude::*;
+use parity_scale_codec::DecodeAll;
 
 #[cfg(test)]
 const TIMEOUT: Duration = Duration::from_millis(10_000);
@@ -34,7 +35,7 @@ where
     /// Unexpected non-binary message received
     NonBinaryMessage,
     /// Error during versioned message decoding
-    IrohaVersion(#[from] iroha_version::error::Error),
+    Decode(#[from] parity_scale_codec::Error),
 }
 
 /// Represents message used by the stream
@@ -56,7 +57,7 @@ pub trait StreamMessage {
 #[async_trait::async_trait]
 pub trait Sink<S>: SinkExt<Self::Message, Error = Self::Err> + Unpin
 where
-    S: EncodeVersioned + Send + Sync + 'static,
+    S: Encode + Send + Sync + 'static,
 {
     /// Error type returned by the sink
     type Err: std::error::Error + Send + Sync + 'static;
@@ -68,10 +69,7 @@ where
     async fn send(&mut self, message: S) -> Result<(), Error<Self::Err>> {
         tokio::time::timeout(
             TIMEOUT,
-            <Self as SinkExt<Self::Message>>::send(
-                self,
-                Self::Message::binary(message.encode_versioned()),
-            ),
+            <Self as SinkExt<Self::Message>>::send(self, Self::Message::binary(message.encode())),
         )
         .await
         .map_err(|_err| Error::SendTimeout)?
@@ -81,7 +79,7 @@ where
 
 /// Trait for reading custom messages from stream
 #[async_trait::async_trait]
-pub trait Stream<R: DecodeVersioned>:
+pub trait Stream<R: DecodeAll>:
     StreamExt<Item = std::result::Result<Self::Message, Self::Err>> + Unpin
 {
     /// Error type returned by the stream
@@ -106,9 +104,7 @@ pub trait Stream<R: DecodeVersioned>:
             return Err(Error::NonBinaryMessage);
         }
 
-        Ok(R::decode_all_versioned(
-            subscription_request_message.as_bytes(),
-        )?)
+        Ok(R::decode_all(&mut subscription_request_message.as_bytes())?)
     }
 }
 
@@ -133,14 +129,14 @@ impl StreamMessage for warp::ws::Message {
 #[async_trait::async_trait]
 impl<M> Sink<M> for warp::ws::WebSocket
 where
-    M: EncodeVersioned + Send + Sync + 'static,
+    M: Encode + Send + Sync + 'static,
 {
     type Err = warp::Error;
     type Message = warp::ws::Message;
 }
 
 #[async_trait::async_trait]
-impl<M: DecodeVersioned> Stream<M> for warp::ws::WebSocket {
+impl<M: DecodeAll> Stream<M> for warp::ws::WebSocket {
     type Err = warp::Error;
     type Message = warp::ws::Message;
 }
@@ -152,14 +148,14 @@ mod ws_client {
     use super::*;
 
     #[async_trait::async_trait]
-    impl<M: DecodeVersioned> Stream<M> for WsClient {
+    impl<M: DecodeAll> Stream<M> for WsClient {
         type Err = warp::test::WsError;
         type Message = warp::ws::Message;
     }
     #[async_trait::async_trait]
     impl<M> Sink<M> for WsClient
     where
-        M: EncodeVersioned + Send + Sync + 'static,
+        M: Encode + Send + Sync + 'static,
     {
         type Err = warp::test::WsError;
         type Message = warp::ws::Message;

cli/src/stream.rs

@@ -25,11 +25,8 @@ use iroha_core::{
 };
 use iroha_data_model::{
     block::{
-        stream::{
-            BlockMessage, BlockSubscriptionRequest, VersionedBlockMessage,
-            VersionedBlockSubscriptionRequest,
-        },
-        VersionedCommittedBlock,
+        stream::{BlockMessage, BlockSubscriptionRequest},
+        VersionedSignedBlock,
     },
     http::{BatchedResponse, VersionedBatchedResponse},
     prelude::*,
@@ -278,8 +275,7 @@ async fn handle_post_configuration(
 
 #[iroha_futures::telemetry_future]
 async fn handle_blocks_stream(kura: Arc<Kura>, mut stream: WebSocket) -> eyre::Result<()> {
-    let subscription_request: VersionedBlockSubscriptionRequest = stream.recv().await?;
-    let BlockSubscriptionRequest(mut from_height) = subscription_request.into_v1();
+    let BlockSubscriptionRequest(mut from_height) = stream.recv().await?;
 
     let mut interval = tokio::time::interval(std::time::Duration::from_millis(10));
     loop {
@@ -307,10 +303,8 @@ async fn handle_blocks_stream(kura: Arc<Kura>, mut stream: WebSocket) -> eyre::R
             _ = interval.tick() => {
                 if let Some(block) = kura.get_block_by_height(from_height.get()) {
                     stream
-                        // TODO: to avoid clone `VersionedBlockMessage` could be split into sending and receiving parts
-                        .send(VersionedBlockMessage::from(
-                            BlockMessage(VersionedCommittedBlock::clone(&block)),
-                        ))
+                        // TODO: to avoid clone `BlockMessage` could be split into sending and receiving parts
+                        .send(BlockMessage(VersionedSignedBlock::clone(&block)))
                         .await?;
                     from_height = from_height.checked_add(1).expect("Maximum block height is achieved.");
                 }

cli/src/torii/routing.rs

@@ -132,14 +132,15 @@ impl Config {
                 let block = blocks
                     .next()
                     .expect("The block is not yet in WSV. Need more sleep?");
-                let block = block.as_v1();
                 (
                     block
+                        .payload()
                         .transactions
                         .iter()
                         .filter(|tx| tx.error.is_none())
                         .count(),
                     block
+                        .payload()
                         .transactions
                         .iter()
                         .filter(|tx| tx.error.is_some())
@@ -173,7 +174,7 @@ impl MeasurerUnit {
         let keypair = iroha_crypto::KeyPair::generate().expect("Failed to generate KeyPair.");
 
         let account_id = account_id(self.name);
-        let alice_id = <Account as Identifiable>::Id::from_str("alice@wonderland")?;
+        let alice_id = AccountId::from_str("alice@wonderland")?;
         let asset_id = asset_id(self.name);
 
         let register_me = RegisterBox::new(Account::new(
@@ -238,8 +239,7 @@ impl MeasurerUnit {
         let submitter = self.client.clone();
         let interval_us_per_tx = self.config.interval_us_per_tx;
         let instructions = self.instructions();
-        let alice_id = <Account as Identifiable>::Id::from_str("alice@wonderland")
-            .expect("Failed to parse account id");
+        let alice_id = AccountId::from_str("alice@wonderland").expect("Failed to parse account id");
 
         let mut nonce = NonZeroU32::new(1).expect("Valid");
 

client/benches/tps/utils.rs

@@ -21,7 +21,7 @@ use http_default::{AsyncWebSocketStream, WebSocketStream};
 use iroha_config::{client::Configuration, torii::uri, GetConfiguration, PostConfiguration};
 use iroha_crypto::{HashOf, KeyPair};
 use iroha_data_model::{
-    block::VersionedCommittedBlock,
+    block::VersionedSignedBlock,
     http::VersionedBatchedResponse,
     isi::Instruction,
     predicate::PredicateBox,
@@ -654,7 +654,7 @@ impl Client {
                     PipelineStatus::Rejected(ref reason) => {
                         return Err(reason.clone().into());
                     }
-                    PipelineStatus::Committed => return Ok(hash.transmute()),
+                    PipelineStatus::Committed => return Ok(hash),
                 }
             }
         }
@@ -1036,7 +1036,7 @@ impl Client {
     pub fn listen_for_blocks(
         &self,
         height: NonZeroU64,
-    ) -> Result<impl Iterator<Item = Result<VersionedCommittedBlock>>> {
+    ) -> Result<impl Iterator<Item = Result<VersionedSignedBlock>>> {
         blocks_api::BlockIterator::new(self.blocks_handler(height)?)
     }
 
@@ -1448,10 +1448,7 @@ pub mod events_api {
                     url,
                 } = self;
 
-                let msg =
-                    VersionedEventSubscriptionRequest::from(EventSubscriptionRequest::new(filter))
-                        .encode_versioned();
-
+                let msg = EventSubscriptionRequest::new(filter).encode();
                 InitData::new(R::new(HttpMethod::GET, url).headers(headers), msg, Events)
             }
         }
@@ -1464,8 +1461,7 @@ pub mod events_api {
             type Event = iroha_data_model::prelude::Event;
 
             fn message(&self, message: Vec<u8>) -> Result<Self::Event> {
-                let event_socket_message =
-                    VersionedEventMessage::decode_all_versioned(&message)?.into_v1();
+                let event_socket_message = EventMessage::decode_all(&mut message.as_slice())?;
                 Ok(event_socket_message.into())
             }
         }
@@ -1532,10 +1528,7 @@ mod blocks_api {
                     url,
                 } = self;
 
-                let msg =
-                    VersionedBlockSubscriptionRequest::from(BlockSubscriptionRequest::new(height))
-                        .encode_versioned();
-
+                let msg = BlockSubscriptionRequest::new(height).encode();
                 InitData::new(R::new(HttpMethod::GET, url).headers(headers), msg, Events)
             }
         }
@@ -1545,11 +1538,10 @@ mod blocks_api {
         pub struct Events;
 
         impl FlowEvents for Events {
-            type Event = iroha_data_model::block::VersionedCommittedBlock;
+            type Event = iroha_data_model::block::VersionedSignedBlock;
 
             fn message(&self, message: Vec<u8>) -> Result<Self::Event> {
-                let block_msg = VersionedBlockMessage::decode_all_versioned(&message)?.into_v1();
-                Ok(block_msg.into())
+                Ok(BlockMessage::decode_all(&mut message.as_slice()).map(Into::into)?)
             }
         }
     }
@@ -1610,7 +1602,7 @@ pub mod asset {
     }
 
     /// Construct a query to get an asset by its id
-    pub fn by_id(asset_id: impl Into<EvaluatesTo<<Asset as Identifiable>::Id>>) -> FindAssetById {
+    pub fn by_id(asset_id: impl Into<EvaluatesTo<AssetId>>) -> FindAssetById {
         FindAssetById::new(asset_id)
     }
 }
@@ -1632,7 +1624,7 @@ pub mod block {
 
     /// Construct a query to find block header by hash
     pub fn header_by_hash(
-        hash: impl Into<EvaluatesTo<HashOf<VersionedCommittedBlock>>>,
+        hash: impl Into<EvaluatesTo<HashOf<VersionedSignedBlock>>>,
     ) -> FindBlockHeaderByHash {
         FindBlockHeaderByHash::new(hash)
     }

client/src/client.rs

@@ -14,7 +14,7 @@ use super::Configuration;
 
 #[test]
 fn client_register_asset_should_add_asset_once_but_not_twice() -> Result<()> {
-    let (_rt, _peer, mut test_client) = <PeerBuilder>::new().with_port(10_620).start_with_runtime();
+    let (_rt, _peer, test_client) = <PeerBuilder>::new().with_port(10_620).start_with_runtime();
     wait_for_genesis_committed(&[test_client.clone()], 0);
 
     // Given
@@ -48,7 +48,7 @@ fn client_register_asset_should_add_asset_once_but_not_twice() -> Result<()> {
 
 #[test]
 fn unregister_asset_should_remove_asset_from_account() -> Result<()> {
-    let (_rt, _peer, mut test_client) = <PeerBuilder>::new().with_port(10_555).start_with_runtime();
+    let (_rt, _peer, test_client) = <PeerBuilder>::new().with_port(10_555).start_with_runtime();
     wait_for_genesis_committed(&[test_client.clone()], 0);
 
     // Given
@@ -87,7 +87,7 @@ fn unregister_asset_should_remove_asset_from_account() -> Result<()> {
 
 #[test]
 fn client_add_asset_quantity_to_existing_asset_should_increase_asset_amount() -> Result<()> {
-    let (_rt, _peer, mut test_client) = <PeerBuilder>::new().with_port(10_000).start_with_runtime();
+    let (_rt, _peer, test_client) = <PeerBuilder>::new().with_port(10_000).start_with_runtime();
     wait_for_genesis_committed(&[test_client.clone()], 0);
 
     // Given
@@ -120,7 +120,7 @@ fn client_add_asset_quantity_to_existing_asset_should_increase_asset_amount() ->
 
 #[test]
 fn client_add_big_asset_quantity_to_existing_asset_should_increase_asset_amount() -> Result<()> {
-    let (_rt, _peer, mut test_client) = <PeerBuilder>::new().with_port(10_510).start_with_runtime();
+    let (_rt, _peer, test_client) = <PeerBuilder>::new().with_port(10_510).start_with_runtime();
     wait_for_genesis_committed(&[test_client.clone()], 0);
 
     // Given
@@ -153,7 +153,7 @@ fn client_add_big_asset_quantity_to_existing_asset_should_increase_asset_amount(
 
 #[test]
 fn client_add_asset_with_decimal_should_increase_asset_amount() -> Result<()> {
-    let (_rt, _peer, mut test_client) = <PeerBuilder>::new().with_port(10_515).start_with_runtime();
+    let (_rt, _peer, test_client) = <PeerBuilder>::new().with_port(10_515).start_with_runtime();
     wait_for_genesis_committed(&[test_client.clone()], 0);
 
     // Given