From 54c020415880582a8f4cd98bba6d722068af06bb Mon Sep 17 00:00:00 2001 From: Renan Santos Date: Fri, 16 Aug 2024 12:24:35 -0300 Subject: [PATCH] fixup! feat: add the MultidappClaimer as an experimental feature for sunodo --- offchain/authority-claimer/src/listener.rs | 156 ++++++++++++++++--- offchain/rollups-events/src/broker/mod.rs | 18 ++- offchain/rollups-events/src/lib.rs | 2 +- offchain/rollups-events/tests/integration.rs | 38 +++-- 4 files changed, 167 insertions(+), 47 deletions(-) diff --git a/offchain/authority-claimer/src/listener.rs b/offchain/authority-claimer/src/listener.rs index 19103244d..87bca4690 100644 --- a/offchain/authority-claimer/src/listener.rs +++ b/offchain/authority-claimer/src/listener.rs @@ -88,6 +88,7 @@ impl BrokerListener for DefaultBrokerListener { pub struct MultidappBrokerListener { broker: Broker, streams: HashMap, // stream => last-claim-id + buffer: HashMap, chain_id: u64, } @@ -102,9 +103,11 @@ impl MultidappBrokerListener { ); let broker = Broker::new(broker_config).await?; let streams = HashMap::new(); + let buffer = HashMap::new(); Ok(Self { broker, streams, + buffer, chain_id, }) } @@ -116,29 +119,66 @@ impl MultidappBrokerListener { async fn update_streams(&mut self) -> Result<(), BrokerListenerError> { let initial_id = INITIAL_ID.to_string(); - let streams: Vec<_> = - self.broker.get_dapps().await.context(BrokerSnafu)?; - tracing::info!("Got the following dapps: {:?}", self.streams); + // Gets the dapps from the broker. + let dapps = self.broker.get_dapps().await.context(BrokerSnafu)?; + if dapps.is_empty() { + return Err(BrokerListenerError::NoApplicationsConfigured); + } + tracing::info!( + "Got the following dapps from key \"{}\": {:?}", + rollups_events::DAPPS_KEY, + dapps + ); - let streams: Vec<_> = streams + // Converts dapps to streams. + let streams: Vec<_> = dapps .into_iter() .map(|dapp_address| { - let dapp_metadata = &DAppMetadata { + RollupsClaimsStream::new(&DAppMetadata { chain_id: self.chain_id, dapp_address, - }; - let stream = RollupsClaimsStream::new(dapp_metadata); - let id = self.streams.get(&stream).unwrap_or(&initial_id); - (stream, id.clone()) + }) }) .collect(); - if streams.is_empty() { - return Err(BrokerListenerError::NoApplicationsConfigured); + + // Removes obsolete dapps from the buffer, if any. + for key in self.buffer.clone().keys() { + if !streams.contains(key) { + self.buffer.remove(key); + } } + // Adds the last consumed ids. + let streams: Vec<_> = streams + .into_iter() + .map(|stream| { + let id = self.streams.get(&stream).unwrap_or(&initial_id); + (stream, id.to_string()) + }) + .collect(); + self.streams = HashMap::from_iter(streams); Ok(()) } + + async fn fill_buffer(&mut self) -> Result<(), BrokerListenerError> { + let streams_and_events: Vec<_> = self + .broker + .consume_blocking_from_multiple_streams(self.streams.clone()) + .await + .context(BrokerSnafu)?; + + for (stream, event) in streams_and_events { + // Updates the last-consumed-id from the stream. + let replaced = self.streams.insert(stream.clone(), event.id); + assert!(replaced.is_some()); + + let replaced = self.buffer.insert(stream, event.payload); + assert!(replaced.is_none()); + } + + Ok(()) + } } #[async_trait] @@ -146,21 +186,17 @@ impl BrokerListener for MultidappBrokerListener { type Error = BrokerListenerError; async fn listen(&mut self) -> Result { - tracing::trace!("Waiting for claim"); - self.update_streams().await?; - let (stream, event) = self - .broker - .consume_blocking_from_multiple_streams(self.streams.clone()) - .await - .context(BrokerSnafu)?; - - // Updates the last-consumed-id from the stream. - let replaced = self.streams.insert(stream.clone(), event.id); - assert!(replaced.is_some()); + tracing::trace!("Waiting for a claim"); + if self.buffer.is_empty() { + self.fill_buffer().await?; + } - Ok(event.payload) + let buffer = self.buffer.clone(); + let (stream, rollups_claim) = buffer.into_iter().next().unwrap(); + self.buffer.remove(&stream); + Ok(rollups_claim) } } @@ -675,4 +711,78 @@ mod tests { broker_listener_thread.await.unwrap(); } + + #[tokio::test] + async fn multidapp_listen_buffer_order() { + let docker = Cli::default(); + let (fixture, mut listener, dapps) = + setup_multidapp_listener(&docker, false).await.unwrap(); + + let mut epochs = vec![0; dapps.len()]; + let indexes = vec![1, 1, 1, 1, 1, 2, 1, 2, 0]; + multidapp_produce_claims(&fixture, &mut epochs, &dapps, &indexes).await; + + let mut buffers = vec![ + vec![0, 1, 2], // + vec![1, 2], + vec![1], + vec![1], + vec![1], + vec![1], + ]; + + for buffer in buffers.iter_mut() { + for _ in 0..buffer.len() { + println!("Buffer: {:?}", buffer); + let result = listener.listen().await; + assert!(result.is_ok(), "{:?}", result.unwrap_err()); + let dapp_address = result.unwrap().dapp_address; + let index = dapps + .iter() + .position(|address| *address == dapp_address) + .unwrap(); + let index = buffer.iter().position(|i| *i == index).unwrap(); + buffer.remove(index); + } + assert!(buffer.is_empty()); + println!("Emptied one of the buffers"); + } + } + + #[tokio::test] + async fn multidapp_listen_buffer_change() { + let docker = Cli::default(); + let (fixture, mut listener, mut dapps) = + setup_multidapp_listener(&docker, false).await.unwrap(); + + let mut epochs = vec![0; dapps.len()]; + let indexes = vec![2, 2, 2, 0, 1, 0]; + multidapp_produce_claims(&fixture, &mut epochs, &dapps, &indexes).await; + + // Removes the last dapp. + assert!(dapps.pop().is_some()); + fixture.set_dapps(dapps.clone()).await; + + let mut buffers = vec![ + vec![0, 1], // + vec![0], + ]; + + for buffer in buffers.iter_mut() { + for _ in 0..buffer.len() { + println!("Buffer: {:?}", buffer); + let result = listener.listen().await; + assert!(result.is_ok(), "{:?}", result.unwrap_err()); + let dapp_address = result.unwrap().dapp_address; + let index = dapps + .iter() + .position(|address| *address == dapp_address) + .unwrap(); + let index = buffer.iter().position(|i| *i == index).unwrap(); + buffer.remove(index); + } + assert!(buffer.is_empty()); + println!("Emptied one of the buffers"); + } + } } diff --git a/offchain/rollups-events/src/broker/mod.rs b/offchain/rollups-events/src/broker/mod.rs index d44d4fb5c..be9118997 100644 --- a/offchain/rollups-events/src/broker/mod.rs +++ b/offchain/rollups-events/src/broker/mod.rs @@ -27,7 +27,7 @@ use crate::Address; pub mod indexer; pub const INITIAL_ID: &str = "0"; -const DAPPS_KEY: &str = "experimental-dapp-addresses-config"; +pub const DAPPS_KEY: &str = "experimental-dapp-addresses-config"; /// The `BrokerConnection` enum implements the `ConnectionLike` trait /// to satisfy the `AsyncCommands` trait bounds. @@ -292,7 +292,7 @@ impl Broker { &mut self, streams: &Vec, last_consumed_ids: &Vec, - ) -> Result<(S, Event), BrokerError> { + ) -> Result)>, BrokerError> { let reply = retry(self.backoff.clone(), || async { let stream_keys: Vec = streams .iter() @@ -318,16 +318,22 @@ impl Broker { return Err(BrokerError::ConsumeTimeout); } - tracing::trace!("checking if any events were received"); + tracing::trace!("getting the consumed events"); + let mut response: Vec<(S, Event)> = vec![]; for mut stream_key in reply.keys { + tracing::trace!("parsing stream key {:?}", stream_key); if let Some(event) = stream_key.ids.pop() { tracing::trace!("parsing received event"); let stream = S::from_key(stream_key.key); let event = event.try_into()?; - return Ok((stream, event)); + response.push((stream, event)); } } - return Err(BrokerError::FailedToConsume); + if response.is_empty() { + Err(BrokerError::FailedToConsume) + } else { + Ok(response) + } } /// Consume the next event from one of the streams. @@ -342,7 +348,7 @@ impl Broker { >( &mut self, streams: HashMap, // streams to last-consumed-ids - ) -> Result<(S, Event), BrokerError> { + ) -> Result)>, BrokerError> { let (streams, last_consumed_ids): (Vec<_>, Vec<_>) = streams.into_iter().map(identity).unzip(); diff --git a/offchain/rollups-events/src/lib.rs b/offchain/rollups-events/src/lib.rs index 8565637bf..b660c7bdd 100644 --- a/offchain/rollups-events/src/lib.rs +++ b/offchain/rollups-events/src/lib.rs @@ -11,7 +11,7 @@ mod rollups_stream; pub use broker::{ indexer, Broker, BrokerCLIConfig, BrokerConfig, BrokerEndpoint, BrokerError, BrokerMultiStream, BrokerStream, Event, RedactedUrl, Url, - INITIAL_ID, + DAPPS_KEY, INITIAL_ID, }; pub use common::{Address, Hash, Payload, ADDRESS_SIZE, HASH_SIZE}; pub use rollups_claims::{RollupsClaim, RollupsClaimsStream}; diff --git a/offchain/rollups-events/tests/integration.rs b/offchain/rollups-events/tests/integration.rs index bf0c3dcc8..3668e978e 100644 --- a/offchain/rollups-events/tests/integration.rs +++ b/offchain/rollups-events/tests/integration.rs @@ -367,26 +367,28 @@ async fn test_it_consumes_from_multiple_streams() { // Consumes all events using the broker struct. let mut counters = HashMap::new(); - for _ in 0..(N * streams.len()) { - let (stream, event) = broker + for _ in 0..streams.len() { + let streams_and_events = broker .consume_blocking_from_multiple_streams(streams.clone()) .await .expect("failed to consume"); - let i = counters - .entry(stream.clone()) - .and_modify(|n| *n += 1) - .or_insert(0) - .clone(); - - // Asserts that the payload is correct. - let expected = format!("{}{}{}", stream.a, stream.b, i); - assert_eq!(expected, event.payload.data); - - // Updates the map of streams with the last consumed id. - let replaced = streams.insert(stream, event.id); - // And asserts that the key from the map was indeed overwritten. - assert!(replaced.is_some()); + for (stream, event) in streams_and_events { + let i = counters + .entry(stream.clone()) + .and_modify(|n| *n += 1) + .or_insert(0) + .clone(); + + // Asserts that the payload is correct. + let expected = format!("{}{}{}", stream.a, stream.b, i); + assert_eq!(expected, event.payload.data); + + // Updates the map of streams with the last consumed id. + let replaced = streams.insert(stream, event.id); + // And asserts that the key from the map was indeed overwritten. + assert!(replaced.is_some()); + } } // Asserts that N events were consumed from each stream. @@ -414,10 +416,12 @@ async fn test_it_consumes_from_multiple_streams() { // Consumes the final event. let marker = Instant::now(); - let (final_stream, _) = broker + let mut streams_and_events = broker .consume_blocking_from_multiple_streams(streams) .await .expect("failed to consume the final event"); + assert_eq!(1, streams_and_events.len()); + let (final_stream, _) = streams_and_events.pop().unwrap(); // Asserts that the main thread blocked for at least WAIT seconds. assert!(marker.elapsed().as_secs() >= WAIT);