[refactor] #4315: split pipeline events

Signed-off-by: Marin Veršić <[email protected]>
hyperledger-iroha · Mar 14, 2024 · 4b628a6 · 4b628a6
1 parent be0e67f
commit 4b628a6
Show file tree

Hide file tree

Showing 23 changed files with 890 additions and 547 deletions.
diff --git a/cli/src/lib.rs b/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));
         match Self::start_telemetry(&logger, &config).await? {
             TelemetryStartStatus::Started => iroha_logger::info!("Telemetry started"),
             TelemetryStartStatus::NotStarted => iroha_logger::warn!("Telemetry not started"),

diff --git a/client/src/client.rs b/client/src/client.rs
@@ -14,7 +14,13 @@ 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, PipelineEventFilterBox, TransactionEventFilter,
+        TransactionStatus,
+    },
+    query::QueryOutputBox,
+};
 use iroha_logger::prelude::*;
 use iroha_telemetry::metrics::Status;
 use iroha_torii_const::uri as torii_uri;
@@ -606,7 +612,7 @@ impl Client {
                 let event_iterator_result = tokio::time::timeout_at(
                     deadline,
                     self.listen_for_events_async(
-                        PipelineEventFilter::new().hash(hash.into()).into(),
+                        TransactionEventFilter::default().with_hash(hash).into(),
                     ),
                 )
                 .await
@@ -633,17 +639,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());
+                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"
         ))

diff --git a/core/benches/blocks/apply_blocks.rs b/core/benches/blocks/apply_blocks.rs
@@ -41,9 +41,10 @@ impl WsvApplyBlocks {
                 .into_iter()
                 .map(|instructions| {
                     let block = create_block(&mut wsv, instructions, account_id.clone(), &key_pair);
-                    wsv.apply_without_execution(&block).map(|()| block)
+                    let _wsv_events = wsv.apply_without_execution(&block);
+                    block
                 })
-                .collect::<Result<Vec<_>, _>>()?
+                .collect::<Vec<_>>()
         };
 
         Ok(Self { wsv, blocks })

diff --git a/core/benches/blocks/common.rs b/core/benches/blocks/common.rs
@@ -42,7 +42,9 @@ pub fn create_block(
     )
     .chain(0, wsv)
     .sign(key_pair)
+    .unpack(|_| {})
     .commit(&topology)
+    .unpack(|_| {})
     .unwrap();
 
     // Verify that transactions are valid

diff --git a/core/benches/blocks/validate_blocks.rs b/core/benches/blocks/validate_blocks.rs
@@ -70,7 +70,7 @@ impl WsvValidateBlocks {
         for (instructions, i) in instructions.into_iter().zip(1..) {
             finalized_wsv = wsv.clone();
             let block = create_block(&mut wsv, instructions, account_id.clone(), &key_pair);
-            wsv.apply_without_execution(&block)?;
+            let _wsv_events = wsv.apply_without_execution(&block);
             assert_eq!(wsv.height(), i);
             assert_eq!(wsv.height(), finalized_wsv.height() + 1);
         }

diff --git a/core/benches/kura.rs b/core/benches/kura.rs
@@ -53,7 +53,8 @@ async fn measure_block_size_for_n_executors(n_executors: u32) {
     let topology = Topology::new(UniqueVec::new());
     let mut block = BlockBuilder::new(vec![tx], topology, Vec::new())
         .chain(0, &mut wsv)
-        .sign(&KeyPair::random());
+        .sign(&KeyPair::random())
+        .unpack(|_| {});
 
     for _ in 1..n_executors {
         block = block.sign(&KeyPair::random());

diff --git a/core/benches/validation.rs b/core/benches/validation.rs
@@ -180,7 +180,7 @@ fn sign_blocks(criterion: &mut Criterion) {
         b.iter_batched(
             || block.clone(),
             |block| {
-                let _: ValidBlock = block.sign(&key_pair);
+                let _: ValidBlock = block.sign(&key_pair).unpack(|_| {});
                 count += 1;
             },
             BatchSize::SmallInput,