[refactor] hyperledger-iroha#4315: split pipeline events

Signed-off-by: Marin Veršić <[email protected]>

cli/src/lib.rs

@@ -251,7 +251,7 @@ impl Iroha {
         });
         let state = Arc::new(state);
 
-        let queue = Arc::new(Queue::from_config(config.queue));
+        let queue = Arc::new(Queue::from_config(config.queue, events_sender.clone()));
         match Self::start_telemetry(&logger, &config).await? {
             TelemetryStartStatus::Started => iroha_logger::info!("Telemetry started"),
             TelemetryStartStatus::NotStarted => iroha_logger::warn!("Telemetry not started"),

client/benches/tps/utils.rs

@@ -18,6 +18,7 @@ use iroha_client::{
         prelude::*,
     },
 };
+use iroha_data_model::events::pipeline::{BlockEventFilter, BlockStatus};
 use serde::Deserialize;
 use test_network::*;
 
@@ -172,9 +173,7 @@ impl MeasurerUnit {
     fn spawn_event_counter(&self) -> thread::JoinHandle<Result<()>> {
         let listener = self.client.clone();
         let (init_sender, init_receiver) = mpsc::channel();
-        let event_filter = PipelineEventFilter::new()
-            .for_entity(PipelineEntityKind::Block)
-            .for_status(PipelineStatusKind::Committed);
+        let event_filter = BlockEventFilter::default().for_status(BlockStatus::Applied);
         let blocks_expected = self.config.blocks as usize;
         let name = self.name;
         let handle = thread::spawn(move || -> Result<()> {

client/src/client.rs

@@ -14,7 +14,10 @@ use eyre::{eyre, Result, WrapErr};
 use futures_util::StreamExt;
 use http_default::{AsyncWebSocketStream, WebSocketStream};
 pub use iroha_config::client_api::ConfigDTO;
-use iroha_data_model::query::QueryOutputBox;
+use iroha_data_model::{
+    events::pipeline::{BlockStatus, PipelineEventBox, TransactionEventFilter, TransactionStatus},
+    query::QueryOutputBox,
+};
 use iroha_logger::prelude::*;
 use iroha_telemetry::metrics::Status;
 use iroha_torii_const::uri as torii_uri;
@@ -605,7 +608,7 @@ impl Client {
             let mut event_iterator = {
                 let event_iterator_result = tokio::time::timeout_at(
                     deadline,
-                    self.listen_for_events_async(PipelineEventFilter::new().for_hash(hash.into())),
+                    self.listen_for_events_async(TransactionEventFilter::default().for_hash(hash)),
                 )
                 .await
                 .map_err(Into::into)
@@ -631,17 +634,34 @@ impl Client {
         event_iterator: &mut AsyncEventStream,
         hash: HashOf<SignedTransaction>,
     ) -> Result<HashOf<SignedTransaction>> {
+        let mut block_height = None;
+
         while let Some(event) = event_iterator.next().await {
-            if let Event::Pipeline(this_event) = event? {
-                match this_event.status() {
-                    PipelineStatus::Validating => {}
-                    PipelineStatus::Rejected(ref reason) => {
-                        return Err(reason.clone().into());
+            if let EventBox::Pipeline(this_event) = event? {
+                match this_event {
+                    PipelineEventBox::Transaction(transaction_event) => {
+                        match transaction_event.status() {
+                            TransactionStatus::Queued => {}
+                            TransactionStatus::Approved => {
+                                block_height = transaction_event.block_height;
+                            }
+                            TransactionStatus::Rejected(reason) => {
+                                return Err(reason.clone().into());
+                            }
+                            TransactionStatus::Expired => return Err(eyre!("Transaction expired")),
+                        }
+                    }
+                    PipelineEventBox::Block(block_event) => {
+                        if Some(block_event.height) == block_height {
+                            if let BlockStatus::Applied = block_event.status() {
+                                return Ok(hash);
+                            }
+                        }
                     }
-                    PipelineStatus::Committed => return Ok(hash),
                 }
             }
         }
+
         Err(eyre!(
             "Connection dropped without `Committed` or `Rejected` event"
         ))
@@ -904,9 +924,7 @@ impl Client {
     pub fn listen_for_events(
         &self,
         event_filter: impl Into<EventFilterBox>,
-    ) -> Result<impl Iterator<Item = Result<Event>>> {
-        let event_filter = event_filter.into();
-        iroha_logger::trace!(?event_filter);
+    ) -> Result<impl Iterator<Item = Result<EventBox>>> {
         events_api::EventIterator::new(self.events_handler(event_filter)?)
     }
 
@@ -919,8 +937,6 @@ impl Client {
         &self,
         event_filter: impl Into<EventFilterBox> + Send,
     ) -> Result<AsyncEventStream> {
-        let event_filter = event_filter.into();
-        iroha_logger::trace!(?event_filter, "Async listening with");
         events_api::AsyncEventStream::new(self.events_handler(event_filter)?).await
     }
 
@@ -933,8 +949,11 @@ impl Client {
         &self,
         event_filter: impl Into<EventFilterBox>,
     ) -> Result<events_api::flow::Init> {
+        let event_filter = event_filter.into();
+        iroha_logger::trace!(?event_filter);
+
         events_api::flow::Init::new(
-            event_filter.into(),
+            event_filter,
             self.headers.clone(),
             self.torii_url
                 .join(torii_uri::SUBSCRIPTION)
@@ -1284,7 +1303,7 @@ pub mod events_api {
         pub struct Events;
 
         impl FlowEvents for Events {
-            type Event = crate::data_model::prelude::Event;
+            type Event = crate::data_model::prelude::EventBox;
 
             fn message(&self, message: Vec<u8>) -> Result<Self::Event> {
                 let event_socket_message = EventMessage::decode_all(&mut message.as_slice())?;

client/src/config.rs

@@ -9,7 +9,7 @@ use iroha_config::{
     base,
     base::{FromEnv, StdEnv, UnwrapPartial},
 };
-use iroha_crypto::prelude::*;
+use iroha_crypto::KeyPair;
 use iroha_data_model::{prelude::*, ChainId};
 use iroha_primitives::small::SmallStr;
 use serde::{Deserialize, Serialize};

client/tests/integration/asset.rs

@@ -10,6 +10,7 @@ use iroha_config::parameters::actual::Root as Config;
 use iroha_data_model::{
     asset::{AssetId, AssetValue, AssetValueType},
     isi::error::{InstructionEvaluationError, InstructionExecutionError, Mismatch, TypeError},
+    transaction::error::TransactionRejectionReason,
 };
 use serde_json::json;
 use test_network::*;
@@ -463,17 +464,17 @@ fn fail_if_dont_satisfy_spec() {
             .expect_err("Should be rejected due to non integer value");
 
         let rejection_reason = err
-            .downcast_ref::<PipelineRejectionReason>()
-            .unwrap_or_else(|| panic!("Error {err} is not PipelineRejectionReason"));
+            .downcast_ref::<TransactionRejectionReason>()
+            .unwrap_or_else(|| panic!("Error {err} is not TransactionRejectionReason"));
 
         assert_eq!(
             rejection_reason,
-            &PipelineRejectionReason::Transaction(TransactionRejectionReason::Validation(
-                ValidationFail::InstructionFailed(InstructionExecutionError::Evaluate(
-                    InstructionEvaluationError::Type(TypeError::from(Mismatch {
+            &TransactionRejectionReason::Validation(ValidationFail::InstructionFailed(
+                InstructionExecutionError::Evaluate(InstructionEvaluationError::Type(
+                    TypeError::from(Mismatch {
                         expected: AssetValueType::Numeric(NumericSpec::integer()),
                         actual: AssetValueType::Numeric(NumericSpec::fractional(2))
-                    }))
+                    })
                 ))
             ))
         );

client/tests/integration/domain_owner_permissions.rs

@@ -3,6 +3,7 @@ use iroha_client::{
     crypto::KeyPair,
     data_model::{account::SignatureCheckCondition, prelude::*},
 };
+use iroha_data_model::transaction::error::TransactionRejectionReason;
 use serde_json::json;
 use test_network::*;
 
@@ -37,14 +38,12 @@ fn domain_owner_domain_permissions() -> Result<()> {
         .expect_err("Tx should fail due to permissions");
 
     let rejection_reason = err
-        .downcast_ref::<PipelineRejectionReason>()
-        .unwrap_or_else(|| panic!("Error {err} is not PipelineRejectionReason"));
+        .downcast_ref::<TransactionRejectionReason>()
+        .unwrap_or_else(|| panic!("Error {err} is not TransactionRejectionReason"));
 
     assert!(matches!(
         rejection_reason,
-        &PipelineRejectionReason::Transaction(TransactionRejectionReason::Validation(
-            ValidationFail::NotPermitted(_)
-        ))
+        &TransactionRejectionReason::Validation(ValidationFail::NotPermitted(_))
     ));
 
     // "alice@wonderland" owns the domain and can register AssetDefinitions by default as domain owner

client/tests/integration/events/pipeline.rs

@@ -9,6 +9,14 @@ use iroha_client::{
     },
 };
 use iroha_config::parameters::actual::Root as Config;
+use iroha_data_model::{
+    events::pipeline::{
+        BlockEvent, BlockEventFilter, BlockStatus, TransactionEventFilter, TransactionStatus,
+    },
+    isi::error::InstructionExecutionError,
+    transaction::error::TransactionRejectionReason,
+    ValidationFail,
+};
 use test_network::*;
 
 // Needed to re-enable ignored tests.
@@ -17,24 +25,28 @@ const PEER_COUNT: usize = 7;
 #[ignore = "ignore, more in #2851"]
 #[test]
 fn transaction_with_no_instructions_should_be_committed() -> Result<()> {
-    test_with_instruction_and_status_and_port(None, PipelineStatusKind::Committed, 10_250)
+    test_with_instruction_and_status_and_port(None, &TransactionStatus::Approved, 10_250)
 }
 
 #[ignore = "ignore, more in #2851"]
 // #[ignore = "Experiment"]
 #[test]
 fn transaction_with_fail_instruction_should_be_rejected() -> Result<()> {
-    let fail = Fail::new("Should be rejected".to_owned());
+    let msg = "Should be rejected".to_owned();
+
+    let fail = Fail::new(msg.clone());
     test_with_instruction_and_status_and_port(
         Some(fail.into()),
-        PipelineStatusKind::Rejected,
+        &TransactionStatus::Rejected(Box::new(TransactionRejectionReason::Validation(
+            ValidationFail::InstructionFailed(InstructionExecutionError::Fail(msg)),
+        ))),
         10_350,
     )
 }
 
 fn test_with_instruction_and_status_and_port(
     instruction: Option<InstructionBox>,
-    should_be: PipelineStatusKind,
+    should_be: &TransactionStatus,
     port: u16,
 ) -> Result<()> {
     let (_rt, network, client) =
@@ -56,9 +68,9 @@ fn test_with_instruction_and_status_and_port(
     let mut handles = Vec::new();
     for listener in clients {
         let checker = Checker { listener, hash };
-        let handle_validating = checker.clone().spawn(PipelineStatusKind::Validating);
+        let handle_validating = checker.clone().spawn(TransactionStatus::Queued);
         handles.push(handle_validating);
-        let handle_validated = checker.spawn(should_be);
+        let handle_validated = checker.spawn(should_be.clone());
         handles.push(handle_validated);
     }
     // When
@@ -78,15 +90,14 @@ struct Checker {
 }
 
 impl Checker {
-    fn spawn(self, status_kind: PipelineStatusKind) -> JoinHandle<()> {
+    fn spawn(self, status_kind: TransactionStatus) -> JoinHandle<()> {
         thread::spawn(move || {
             let mut event_iterator = self
                 .listener
                 .listen_for_events(
-                    PipelineEventFilter::new()
-                        .for_entity(PipelineEntityKind::Transaction)
+                    TransactionEventFilter::default()
                         .for_status(status_kind)
-                        .for_hash(*self.hash),
+                        .for_hash(self.hash),
                 )
                 .expect("Failed to create event iterator.");
             let event_result = event_iterator.next().expect("Stream closed");
@@ -96,13 +107,11 @@ impl Checker {
 }
 
 #[test]
-fn committed_block_must_be_available_in_kura() {
+fn applied_block_must_be_available_in_kura() {
     let (_rt, peer, client) = <PeerBuilder>::new().with_port(11_040).start_with_runtime();
     wait_for_genesis_committed(&[client.clone()], 0);
 
-    let event_filter = PipelineEventFilter::new()
-        .for_entity(PipelineEntityKind::Block)
-        .for_status(PipelineStatusKind::Committed);
+    let event_filter = BlockEventFilter::default().for_status(BlockStatus::Committed);
     let mut event_iter = client
         .listen_for_events(event_filter)
         .expect("Failed to subscribe for events");
@@ -111,21 +120,17 @@ fn committed_block_must_be_available_in_kura() {
         .submit(Fail::new("Dummy instruction".to_owned()))
         .expect("Failed to submit transaction");
 
-    let event = event_iter.next().expect("Block must be committed");
-    let Ok(Event::Pipeline(PipelineEvent {
-        entity_kind: PipelineEntityKind::Block,
-        status: PipelineStatus::Committed,
-        hash,
-    })) = event
-    else {
-        panic!("Received unexpected event")
-    };
-    let hash = HashOf::from_untyped_unchecked(hash);
+    let event: BlockEvent = event_iter
+        .next()
+        .expect("Block must be committed")
+        .expect("Block must be committed")
+        .try_into()
+        .expect("Received unexpected event");
 
     peer.iroha
         .as_ref()
         .expect("Must be some")
         .kura
-        .get_block_height_by_hash(&hash)
+        .get_block_by_height(event.height)
         .expect("Block committed event was received earlier");
 }

client/tests/integration/permissions.rs

@@ -6,7 +6,9 @@ use iroha_client::{
     crypto::KeyPair,
     data_model::prelude::*,
 };
-use iroha_data_model::permission::PermissionToken;
+use iroha_data_model::{
+    permission::PermissionToken, transaction::error::TransactionRejectionReason,
+};
 use iroha_genesis::GenesisNetwork;
 use serde_json::json;
 use test_network::{PeerBuilder, *};
@@ -104,14 +106,12 @@ fn permissions_disallow_asset_transfer() {
         .submit_transaction_blocking(&transfer_tx)
         .expect_err("Transaction was not rejected.");
     let rejection_reason = err
-        .downcast_ref::<PipelineRejectionReason>()
-        .expect("Error {err} is not PipelineRejectionReason");
+        .downcast_ref::<TransactionRejectionReason>()
+        .expect("Error {err} is not TransactionRejectionReason");
     //Then
     assert!(matches!(
         rejection_reason,
-        &PipelineRejectionReason::Transaction(TransactionRejectionReason::Validation(
-            ValidationFail::NotPermitted(_)
-        ))
+        &TransactionRejectionReason::Validation(ValidationFail::NotPermitted(_))
     ));
     let alice_assets = get_assets(&iroha_client, &alice_id);
     assert_eq!(alice_assets, alice_start_assets);
@@ -156,14 +156,12 @@ fn permissions_disallow_asset_burn() {
         .submit_transaction_blocking(&burn_tx)
         .expect_err("Transaction was not rejected.");
     let rejection_reason = err
-        .downcast_ref::<PipelineRejectionReason>()
-        .expect("Error {err} is not PipelineRejectionReason");
+        .downcast_ref::<TransactionRejectionReason>()
+        .expect("Error {err} is not TransactionRejectionReason");
 
     assert!(matches!(
         rejection_reason,
-        &PipelineRejectionReason::Transaction(TransactionRejectionReason::Validation(
-            ValidationFail::NotPermitted(_)
-        ))
+        &TransactionRejectionReason::Validation(ValidationFail::NotPermitted(_))
     ));
 
     let alice_assets = get_assets(&iroha_client, &alice_id);

client/tests/integration/roles.rs

@@ -6,6 +6,7 @@ use iroha_client::{
     crypto::KeyPair,
     data_model::prelude::*,
 };
+use iroha_data_model::transaction::error::TransactionRejectionReason;
 use serde_json::json;
 use test_network::*;
 
@@ -164,14 +165,12 @@ fn role_with_invalid_permissions_is_not_accepted() -> Result<()> {
         .expect_err("Submitting role with invalid permission token should fail");
 
     let rejection_reason = err
-        .downcast_ref::<PipelineRejectionReason>()
-        .unwrap_or_else(|| panic!("Error {err} is not PipelineRejectionReason"));
+        .downcast_ref::<TransactionRejectionReason>()
+        .unwrap_or_else(|| panic!("Error {err} is not TransactionRejectionReason"));
 
     assert!(matches!(
         rejection_reason,
-        &PipelineRejectionReason::Transaction(TransactionRejectionReason::Validation(
-            ValidationFail::NotPermitted(_)
-        ))
+        &TransactionRejectionReason::Validation(ValidationFail::NotPermitted(_))
     ));
 
     Ok(())