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

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

cli/src/lib.rs

@@ -240,7 +240,7 @@ impl Iroha {
             },
         );
 
-        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::*;
 
@@ -171,10 +172,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()
-            .entity_kind(PipelineEntityKind::Block)
-            .status_kind(PipelineStatusKind::Committed)
-            .into();
+        let event_filter = BlockEventFilter::default().with_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,9 +608,7 @@ impl Client {
             let mut event_iterator = {
                 let event_iterator_result = tokio::time::timeout_at(
                     deadline,
-                    self.listen_for_events_async(
-                        PipelineEventFilter::new().hash(hash.into()).into(),
-                    ),
+                    self.listen_for_events_async(TransactionEventFilter::default().with_hash(hash)),
                 )
                 .await
                 .map_err(Into::into)
@@ -633,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"
         ))
@@ -905,9 +923,8 @@ impl Client {
     /// - Forwards from [`events_api::EventIterator::new`]
     pub fn listen_for_events(
         &self,
-        event_filter: FilterBox,
-    ) -> Result<impl Iterator<Item = Result<Event>>> {
-        iroha_logger::trace!(?event_filter);
+        event_filter: impl Into<FilterBox>,
+    ) -> Result<impl Iterator<Item = Result<EventBox>>> {
         events_api::EventIterator::new(self.events_handler(event_filter)?)
     }
 
@@ -918,9 +935,8 @@ impl Client {
     /// - Forwards from [`events_api::AsyncEventStream::new`]
     pub async fn listen_for_events_async(
         &self,
-        event_filter: FilterBox,
+        event_filter: impl Into<FilterBox> + Send,
     ) -> Result<AsyncEventStream> {
-        iroha_logger::trace!(?event_filter, "Async listening with");
         events_api::AsyncEventStream::new(self.events_handler(event_filter)?).await
     }
 
@@ -929,7 +945,13 @@ impl Client {
     /// # Errors
     /// Fails if handler construction fails
     #[inline]
-    pub fn events_handler(&self, event_filter: FilterBox) -> Result<events_api::flow::Init> {
+    pub fn events_handler(
+        &self,
+        event_filter: impl Into<FilterBox>,
+    ) -> Result<events_api::flow::Init> {
+        let event_filter = event_filter.into();
+        iroha_logger::trace!(?event_filter);
+
         events_api::flow::Init::new(
             event_filter,
             self.headers.clone(),
@@ -1281,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/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/data.rs

@@ -138,7 +138,7 @@ fn transaction_execution_should_produce_events(
     let listener = client.clone();
     let (init_sender, init_receiver) = mpsc::channel();
     let (event_sender, event_receiver) = mpsc::channel();
-    let event_filter = DataEventFilter::AcceptAll.into();
+    let event_filter = DataEventFilter::AcceptAll;
     thread::spawn(move || -> Result<()> {
         let event_iterator = listener.listen_for_events(event_filter)?;
         init_sender.send(())?;
@@ -182,7 +182,7 @@ fn produce_multiple_events() -> Result<()> {
     let listener = client.clone();
     let (init_sender, init_receiver) = mpsc::channel();
     let (event_sender, event_receiver) = mpsc::channel();
-    let event_filter = DataEventFilter::AcceptAll.into();
+    let event_filter = DataEventFilter::AcceptAll;
     thread::spawn(move || -> Result<()> {
         let event_iterator = listener.listen_for_events(event_filter)?;
         init_sender.send(())?;

client/tests/integration/events/notification.rs

@@ -38,8 +38,7 @@ fn trigger_completion_success_should_produce_event() -> Result<()> {
             NotificationEventFilter::TriggerCompleted(TriggerCompletedEventFilter::new(
                 Some(trigger_id),
                 Some(TriggerCompletedOutcomeType::Success),
-            ))
-            .into(),
+            )),
         )?;
         if event_it.next().is_some() {
             sender.send(())?;
@@ -87,8 +86,7 @@ fn trigger_completion_failure_should_produce_event() -> Result<()> {
             NotificationEventFilter::TriggerCompleted(TriggerCompletedEventFilter::new(
                 Some(trigger_id),
                 Some(TriggerCompletedOutcomeType::Failure),
-            ))
-            .into(),
+            )),
         )?;
         if event_it.next().is_some() {
             sender.send(())?;

client/tests/integration/events/pipeline.rs

@@ -9,6 +9,9 @@ use iroha_client::{
     },
 };
 use iroha_config::parameters::actual::Root as Config;
+use iroha_data_model::events::pipeline::{
+    BlockEvent, BlockEventFilter, BlockStatus, TransactionEventFilter, TransactionStatus,
+};
 use test_network::*;
 
 // Needed to re-enable ignored tests.
@@ -17,14 +20,15 @@ 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::Committed, 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());
+
     test_with_instruction_and_status_and_port(
         Some(fail.into()),
         PipelineStatusKind::Rejected,
@@ -34,7 +38,7 @@ fn transaction_with_fail_instruction_should_be_rejected() -> Result<()> {
 
 fn test_with_instruction_and_status_and_port(
     instruction: Option<InstructionBox>,
-    should_be: PipelineStatusKind,
+    should_be: TransactionStatus,
     port: u16,
 ) -> Result<()> {
     let (_rt, network, client) =
@@ -78,16 +82,15 @@ 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(FilterBox::Pipeline(
-                    PipelineEventFilter::new()
-                        .entity_kind(PipelineEntityKind::Transaction)
-                        .status_kind(status_kind)
-                        .hash(*self.hash),
-                ))
+                .listen_for_events(
+                    TransactionEventFilter::default()
+                        .with_status(status_kind)
+                        .with_hash(self.hash),
+                )
                 .expect("Failed to create event iterator.");
             let event_result = event_iterator.next().expect("Stream closed");
             let _event = event_result.expect("Must be valid");
@@ -96,14 +99,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()
-        .entity_kind(PipelineEntityKind::Block)
-        .status_kind(PipelineStatusKind::Committed)
-        .into();
+    let event_filter = BlockEventFilter::default().with_status(BlockStatus::Committed);
     let mut event_iter = client
         .listen_for_events(event_filter)
         .expect("Failed to subscribe for events");
@@ -113,20 +113,18 @@ fn committed_block_must_be_available_in_kura() {
         .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
+    let Ok(EventBox::Pipeline(PipelineEventBox::Block(BlockEvent {
+        status: BlockStatus::Applied,
+        height,
+    }))) = event
     else {
         panic!("Received unexpected event")
     };
-    let hash = HashOf::from_untyped_unchecked(hash);
 
     peer.iroha
         .as_ref()
         .expect("Must be some")
         .kura
-        .get_block_height_by_hash(&hash)
+        .get_block_by_height(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);