Modify Relayer service to order Events from L1 by block index

crates/services/relayer/src/mock_db.rs

@@ -16,10 +16,7 @@ use fuel_core_types::{
     services::relayer::Event,
 };
 use std::{
-    collections::{
-        BTreeMap,
-        HashMap,
-    },
+    collections::BTreeMap,
     sync::{
         Arc,
         Mutex,
@@ -28,9 +25,9 @@ use std::{
 
 #[derive(Default)]
 pub struct Data {
-    pub messages: BTreeMap<DaBlockHeight, HashMap<Nonce, Message>>,
+    pub messages: BTreeMap<DaBlockHeight, Vec<(Nonce, Message)>>,
     pub transactions:
-        BTreeMap<DaBlockHeight, HashMap<RelayedTransactionId, RelayedTransaction>>,
+        BTreeMap<DaBlockHeight, Vec<(RelayedTransactionId, RelayedTransaction)>>,
     pub finalized_da_height: Option<DaBlockHeight>,
 }
 
@@ -44,13 +41,27 @@ pub struct MockDb {
 }
 
 impl MockDb {
-    pub fn get_message(&self, id: &Nonce) -> Option<Message> {
+    pub fn get_message(&self, nonce: &Nonce) -> Option<Message> {
         self.data
             .lock()
             .unwrap()
             .messages
             .iter()
-            .find_map(|(_, map)| map.get(id).cloned())
+            .find_map(|(_, map)| {
+                map.iter()
+                    .find(|(inner_nonce, _msg)| nonce == inner_nonce)
+                    .map(|(_, msg)| msg.clone())
+            })
+    }
+
+    pub fn get_messages_for_block(&self, da_block_height: DaBlockHeight) -> Vec<Message> {
+        self.data
+            .lock()
+            .unwrap()
+            .messages
+            .get(&da_block_height)
+            .map(|map| map.iter().map(|(_, msg)| msg).cloned().collect())
+            .unwrap_or_default()
     }
 
     pub fn get_transaction(
@@ -62,7 +73,24 @@ impl MockDb {
             .unwrap()
             .transactions
             .iter()
-            .find_map(|(_, map)| map.get(id).cloned())
+            .find_map(|(_, txs)| {
+                txs.iter()
+                    .find(|(inner_id, _tx)| id == inner_id)
+                    .map(|(_, tx)| tx.clone())
+            })
+    }
+
+    pub fn get_transactions_for_block(
+        &self,
+        da_block_height: DaBlockHeight,
+    ) -> Vec<RelayedTransaction> {
+        self.data
+            .lock()
+            .unwrap()
+            .transactions
+            .get(&da_block_height)
+            .map(|map| map.iter().map(|(_, tx)| tx).cloned().collect())
+            .unwrap_or_default()
     }
 }
 
@@ -79,13 +107,13 @@ impl RelayerDb for MockDb {
                     m.messages
                         .entry(message.da_height())
                         .or_default()
-                        .insert(*message.id(), message.clone());
+                        .push((*message.id(), message.clone()));
                 }
                 Event::Transaction(transaction) => {
                     m.transactions
                         .entry(transaction.da_height())
                         .or_default()
-                        .insert(transaction.id(), transaction.clone());
+                        .push((transaction.id(), transaction.clone()));
                 }
             }
         }

crates/services/relayer/src/service.rs

@@ -153,6 +153,7 @@ where
             self.config.log_page_size,
         );
         let logs = logs.take_until(self.shutdown.while_started());
+
         write_logs(&mut self.database, logs).await
     }
 

crates/services/relayer/src/service/get_logs.rs

@@ -68,24 +68,24 @@ where
     while let Some((last_height, events)) = logs.try_next().await? {
         let last_height = last_height.into();
         let mut ordered_events = BTreeMap::<DaBlockHeight, Vec<Event>>::new();
-        let fuel_events =
-            events
-                .into_iter()
-                .filter_map(|event| match EthEventLog::try_from(&event) {
-                    Ok(event) => {
-                        match event {
-                            EthEventLog::Message(m) => {
-                                Some(Ok(Event::Message(Message::from(&m))))
-                            }
-                            EthEventLog::Transaction(tx) => {
-                                Some(Ok(Event::Transaction(RelayedTransaction::from(tx))))
-                            }
-                            // TODO: Log out ignored messages.
-                            EthEventLog::Ignored => None,
+        let sorted_events = sort_events_by_log_index(events)?;
+        let fuel_events = sorted_events.into_iter().filter_map(|event| {
+            match EthEventLog::try_from(&event) {
+                Ok(event) => {
+                    match event {
+                        EthEventLog::Message(m) => {
+                            Some(Ok(Event::Message(Message::from(&m))))
                         }
+                        EthEventLog::Transaction(tx) => {
+                            Some(Ok(Event::Transaction(RelayedTransaction::from(tx))))
+                        }
+                        // TODO: Log out ignored messages.
+                        EthEventLog::Ignored => None,
                     }
-                    Err(e) => Some(Err(e)),
-                });
+                }
+                Err(e) => Some(Err(e)),
+            }
+        });
 
         for event in fuel_events {
             let event = event?;
@@ -109,3 +109,18 @@ where
     }
     Ok(())
 }
+
+fn sort_events_by_log_index(events: Vec<Log>) -> anyhow::Result<Vec<Log>> {
+    let mut with_indexes = events
+        .into_iter()
+        .map(|e| {
+            let log_index = e
+                .log_index
+                .ok_or(anyhow::anyhow!("Log missing `log_index`: {e:?}"))?;
+            Ok((log_index, e))
+        })
+        .collect::<anyhow::Result<Vec<_>>>()?;
+    with_indexes.sort_by(|(index_a, _a), (index_b, _b)| index_a.cmp(index_b));
+    let new_events = with_indexes.into_iter().map(|(_, e)| e).collect();
+    Ok(new_events)
+}

crates/services/relayer/src/service/get_logs/test.rs

@@ -119,7 +119,7 @@ async fn can_paginate_logs(input: Input) -> Expected {
     let eth_node = MockMiddleware::default();
 
     eth_node.update_data(|data| {
-        data.logs_batch = vec![logs];
+        data.logs_batch = vec![logs].into();
         data.best_block.number = Some((*eth_gap.end()).into());
     });
     let count = std::sync::Arc::new(AtomicUsize::new(0));

crates/services/relayer/src/service/test.rs

@@ -20,7 +20,7 @@ async fn can_download_logs() {
             ..Default::default()
         },
     ];
-    eth_node.update_data(|data| data.logs_batch = vec![logs.clone()]);
+    eth_node.update_data(|data| data.logs_batch = vec![logs.clone()].into());
 
     let eth_state = super::state::test_builder::TestDataSource {
         eth_remote_finalized: 5,

crates/services/relayer/src/test_helpers/middleware.rs

@@ -51,7 +51,7 @@ struct InnerState {
 pub struct MockData {
     pub is_syncing: SyncingStatus,
     pub best_block: Block<TxHash>,
-    pub logs_batch: Vec<Vec<Log>>,
+    pub logs_batch: Option<Vec<Vec<Log>>>,
     pub logs_batch_index: usize,
 }
 
@@ -141,7 +141,7 @@ impl Default for MockData {
         MockData {
             best_block,
             is_syncing: SyncingStatus::IsFalse,
-            logs_batch: Vec::new(),
+            logs_batch: Some(Vec::new()),
             logs_batch_index: 0,
         }
     }
@@ -263,6 +263,11 @@ impl Middleware for MockMiddleware {
         self.before_event(TriggerType::GetLogs(filter));
         let r = self.update_data(|data| {
             data.logs_batch
+                .take()
+                .unwrap_or_else(|| {
+                    tracing::debug!("No logs batch found");
+                    Vec::new()
+                })
                 .iter()
                 .flat_map(|logs| {
                     logs.iter().filter_map(|log| {