Event monitor: Batch together all events from all transactions included in a block

CHANGELOG.md

@@ -2,7 +2,24 @@
 
 ## Unreleased
 
-> Nothing yet,
+### FEATURES
+
+> Nothing
+
+### IMPROVEMENTS
+
+- [ibc-relayer]
+  - Bulk events from all transactions included in a block ([#957])
+
+### BUG FIXES
+
+> Nothing
+
+### BREAKING CHANGES
+
+> Nothing
+
+[#957]: https://github.com/informalsystems/ibc-rs/issues/957
 
 ## v0.3.1
 *May 14h, 2021*
@@ -32,6 +49,8 @@ as well as support Protobuf-encoded keys.
 
 ### BREAKING CHANGES
 
+> Nothing
+
 
 [#875]: https://github.com/informalsystems/ibc-rs/issues/875
 [#920]: https://github.com/informalsystems/ibc-rs/issues/920

relayer/Cargo.toml

@@ -51,6 +51,7 @@ tonic = "0.4"
 dirs-next = "2.0.0"
 dyn-clone = "1.0.3"
 retry = { version = "1.2.1", default-features = false }
+async-stream = "0.3.1"
 
 [dependencies.tendermint]
 version = "=0.19.0"

relayer/src/event/monitor.rs

@@ -1,13 +1,15 @@
 use std::sync::Arc;
 
 use crossbeam_channel as channel;
-use futures::stream::StreamExt;
-use futures::{stream::select_all, Stream};
-use itertools::Itertools;
+use futures::{
+    pin_mut,
+    stream::{self, select_all, StreamExt},
+    Stream,
+};
 use thiserror::Error;
 use tokio::task::JoinHandle;
 use tokio::{runtime::Runtime as TokioRuntime, sync::mpsc};
-use tracing::{error, info, trace};
+use tracing::{debug, error, info, trace};
 
 use tendermint_rpc::{
     event::Event as RpcEvent,
@@ -18,7 +20,10 @@ use tendermint_rpc::{
 
 use ibc::{events::IbcEvent, ics02_client::height::Height, ics24_host::identifier::ChainId};
 
-use crate::util::retry::{retry_with_index, RetryResult};
+use crate::util::{
+    retry::{retry_with_index, RetryResult},
+    stream::group_while,
+};
 
 mod retry_strategy {
     use crate::util::retry::clamp_total;
@@ -276,24 +281,30 @@ impl EventMonitor {
 
     /// Event monitor loop
     pub fn run(mut self) {
-        info!(chain.id = %self.chain_id, "starting event monitor");
+        debug!(chain.id = %self.chain_id, "starting event monitor");
 
         let rt = self.rt.clone();
 
+        // Take ownership of the subscriptions
+        let subscriptions =
+            std::mem::replace(&mut self.subscriptions, Box::new(futures::stream::empty()));
+
+        // Convert the stream of RPC events into a stream of event batches.
+        let batches = stream_batches(subscriptions, self.chain_id.clone());
+
+        // Needed to be able to poll the stream
+        pin_mut!(batches);
+
         loop {
             let result = rt.block_on(async {
                 tokio::select! {
-                    Some(event) = self.subscriptions.next() => {
-                        event
-                            .map_err(Error::NextEventBatchFailed)
-                            .and_then(|e| self.collect_events(e))
-                    },
+                    Some(batch) = batches.next() => Ok(batch),
                     Some(e) = self.rx_err.recv() => Err(Error::WebSocketDriver(e)),
                 }
             });
 
             match result {
-                Ok(batches) => self.process_batches(batches).unwrap_or_else(|e| {
+                Ok(batch) => self.process_batch(batch).unwrap_or_else(|e| {
                     error!("failed to process event batch: {}", e);
                 }),
                 Err(e) => {
@@ -307,31 +318,43 @@ impl EventMonitor {
     }
 
     /// Collect the IBC events from the subscriptions
-    fn process_batches(&self, batches: Vec<EventBatch>) -> Result<()> {
-        for batch in batches {
-            self.tx_batch
-                .send(Ok(batch))
-                .map_err(|_| Error::ChannelSendFailed)?;
-        }
+    fn process_batch(&self, batch: EventBatch) -> Result<()> {
+        self.tx_batch
+            .send(Ok(batch))
+            .map_err(|_| Error::ChannelSendFailed)?;
 
         Ok(())
     }
+}
 
-    /// Collect the IBC events from the subscriptions
-    fn collect_events(&mut self, event: RpcEvent) -> Result<Vec<EventBatch>> {
-        let ibc_events = crate::event::rpc::get_all_events(&self.chain_id, event)
-            .map_err(Error::CollectEventsFailed)?;
-
-        let events_by_height = ibc_events.into_iter().into_group_map();
-        let batches = events_by_height
-            .into_iter()
-            .map(|(height, events)| EventBatch {
-                chain_id: self.chain_id.clone(),
-                height,
-                events,
-            })
-            .collect();
-
-        Ok(batches)
-    }
+/// Collect the IBC events from an RPC event
+fn collect_events(chain_id: &ChainId, event: RpcEvent) -> impl Stream<Item = (Height, IbcEvent)> {
+    let events = crate::event::rpc::get_all_events(chain_id, event).unwrap_or_default();
+    stream::iter(events)
+}
+
+/// Convert a stream of RPC event into a stream of event batches
+fn stream_batches(
+    subscriptions: Box<SubscriptionStream>,
+    chain_id: ChainId,
+) -> impl Stream<Item = EventBatch> {
+    let id = chain_id.clone();
+
+    // Collect IBC events from each RPC event
+    let batches = subscriptions
+        .filter_map(|rpc_event| async { rpc_event.ok() })
+        .flat_map(move |rpc_event| collect_events(&id, rpc_event));
+
+    // Group events by height
+    let grouped = group_while(batches, |(h0, _), (h1, _)| h0 == h1);
+
+    // Convert each group to a batch
+    grouped.map(move |events| {
+        let (height, _) = events.first().expect("internal error: found empty group"); // SAFETY: upheld by `group_while`
+        EventBatch {
+            height: *height,
+            chain_id: chain_id.clone(),
+            events: events.into_iter().map(|e| e.1).collect(),
+        }
+    })
 }

relayer/src/util.rs

@@ -7,3 +7,4 @@ pub use recv_multiple::{recv_multiple, try_recv_multiple};
 pub mod iter;
 pub mod retry;
 pub mod sled;
+pub mod stream;

relayer/src/util/stream.rs

@@ -0,0 +1,74 @@
+use async_stream::stream;
+use futures::stream::Stream;
+
+/// ## Example
+///
+/// ```rust,ignore
+/// let input = stream::iter(vec![0, 0, 0, 1, 1, 2, 3, 3, 3, 3]);
+/// let output = group_while(stream, |a, b| a == b).collect::<Vec<_>>().await;
+/// assert_eq!(output, vec![vec![0, 0, 0], vec![1, 1], vec![2], vec![3, 3, 3, 3]]);
+/// ```
+pub fn group_while<A, S, F>(input: S, group_these: F) -> impl Stream<Item = Vec<A>>
+where
+    S: Stream<Item = A>,
+    F: Fn(&A, &A) -> bool + 'static,
+{
+    struct State<A> {
+        cur: A,
+        group: Vec<A>,
+    }
+
+    stream! {
+        let mut state = None;
+
+        for await x in input {
+            match &mut state {
+                None => {
+                    state = Some(State { cur: x, group: vec![] });
+                },
+                Some(state) if group_these(&state.cur, &x) => {
+                    let prev = std::mem::replace(&mut state.cur, x);
+                    state.group.push(prev);
+                },
+                Some(state) => {
+                    let cur = std::mem::replace(&mut state.cur, x);
+                    state.group.push(cur);
+                    let group = std::mem::replace(&mut state.group, vec![]);
+                    yield group;
+                }
+            }
+        }
+
+        if let Some(State{ cur, mut group }) = state {
+            group.push(cur);
+            yield group;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::group_while;
+    use futures::{executor::block_on, stream, StreamExt};
+
+    #[test]
+    fn group_while_non_empty() {
+        let input = stream::iter(vec![1, 1, 2, 3, 3, 3, 4, 5, 5]);
+        let output = group_while(input, |a, b| a == b).collect::<Vec<_>>();
+        let result = block_on(output);
+
+        assert_eq!(
+            result,
+            vec![vec![1, 1], vec![2], vec![3, 3, 3], vec![4], vec![5, 5]]
+        );
+    }
+
+    #[test]
+    fn group_while_empty() {
+        let input = stream::iter(Vec::<i32>::new());
+        let output = group_while(input, |a, b| a == b).collect::<Vec<_>>();
+        let result = block_on(output);
+
+        assert_eq!(result, Vec::<Vec<i32>>::new());
+    }
+}