fix(relayer): Accept http requests in gear-eth (#806)

Author: gshep

Cargo.lock

@@ -44,7 +44,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "1.0.0"
+version = "1.0.1"
 edition = "2021"
 
 [workspace.dependencies]

Cargo.toml

@@ -48,7 +48,7 @@ impl BlockListener {
     }
 
     pub async fn run<const RECEIVER_COUNT: usize>(
-        self,
+        mut self,
     ) -> [broadcast::Receiver<GearBlock>; RECEIVER_COUNT] {
         // Capacity for the channel. At the moment merkle-root relayer might lag behind
         // during proof generation or era sync, so we need to have enough capacity
@@ -57,14 +57,13 @@ impl BlockListener {
         let (tx, _) = broadcast::channel(CAPACITY);
         let tx2 = tx.clone();
         tokio::task::spawn(async move {
-            let api = self.api_provider.client();
             let UnprocessedBlocks {
                 last_block,
                 first_block,
                 blocks: _,
             } = self.block_storage.unprocessed_blocks().await;
 
-            let api_back = api.clone();
+            let api_back = self.api_provider.client();
             let tx_back = tx2.clone();
             let storage_back = self.block_storage.clone();
 
@@ -118,9 +117,10 @@ impl BlockListener {
                 Ok(())
             });
 
+            let mut last_finalized_block_number = None;
             loop {
-                let res = self.run_inner(&tx2).await;
-                match res {
+                let res = self.run_inner(&tx2, &mut last_finalized_block_number).await;
+                let e = match res {
                     Ok(false) => {
                         log::info!("Gear block listener stopped due to no active receivers");
                         return;
@@ -131,12 +131,17 @@ impl BlockListener {
                         continue;
                     }
 
-                    Err(err) => {
-                        log::error!("Gear block listener failed: {err}");
+                    Err(e) => e,
+                };
 
-                        return;
-                    }
+                log::error!(r#"Gear block listener failed: "{e:?}""#);
+
+                if let Err(e) = self.api_provider.reconnect().await {
+                    log::error!(r#"API provider unable to reconnect: "{e}""#);
+                    return;
                 }
+
+                log::debug!("API provider reconnected");
             }
         });
 
@@ -147,20 +152,22 @@ impl BlockListener {
             .expect("expected Vec of correct length")
     }
 
-    async fn run_inner(&self, tx: &broadcast::Sender<GearBlock>) -> anyhow::Result<bool> {
+    async fn run_inner(
+        &self,
+        tx: &broadcast::Sender<GearBlock>,
+        last_finalized_block_number: &mut Option<u32>,
+    ) -> anyhow::Result<bool> {
         let gear_api = self.api_provider.client();
 
-        let mut last_finalized_block_number = None;
         let mut subscription = gear_api.subscribe_grandpa_justifications().await?;
-
         while let Some(justification) = subscription.next().await {
             let justification = justification?;
 
             let block_hash = justification.commit.target_hash;
             let block_number = justification.commit.target_number;
 
             // Check if there are missing blocks and fetch them
-            if let Some(last_finalized) = last_finalized_block_number {
+            if let Some(last_finalized) = *last_finalized_block_number {
                 if last_finalized + 1 != block_number {
                     log::info!("Detected gap: last finalized block was #{last_finalized}, current block is #{block_number}");
 
@@ -211,7 +218,7 @@ impl BlockListener {
             }
 
             // Update the last finalized block number
-            last_finalized_block_number = Some(block_number);
+            *last_finalized_block_number = Some(block_number);
             self.metrics.latest_block.set(block_number as i64);
         }
 

relayer/src/message_relayer/common/gear/block_listener.rs

@@ -8,23 +8,27 @@ use anyhow::Result as AnyResult;
 use ethereum_common::U256;
 use std::{cmp::Ordering, ops::Deref};
 use tokio::{
-    sync::mpsc::{UnboundedReceiver, UnboundedSender},
+    sync::mpsc::{UnboundedReceiver, WeakUnboundedSender},
     task,
 };
 
 pub type BlockData = (GearBlock, AuthoritySetId);
 
+/// The purpose of the entity is to receive requests to relay message(s) from an external source,
+/// than get required data for relaying the message(s) and send the data further.
+///
+/// An example of an external requester is HTTP server.
 pub struct MessageDataExtractor {
     api_provider: ApiProviderConnection,
-    sender: UnboundedSender<MessageInBlock>,
+    sender: WeakUnboundedSender<MessageInBlock>,
     receiver: UnboundedReceiver<Message>,
     blocks: BlockDataList,
 }
 
 impl MessageDataExtractor {
     pub fn new(
         api_provider: ApiProviderConnection,
-        sender: UnboundedSender<MessageInBlock>,
+        sender: WeakUnboundedSender<MessageInBlock>,
         receiver: UnboundedReceiver<Message>,
     ) -> Self {
         Self {
@@ -35,33 +39,10 @@ impl MessageDataExtractor {
         }
     }
 
-    pub fn sender(&self) -> &UnboundedSender<MessageInBlock> {
-        &self.sender
-    }
-
     pub fn spawn(self) {
         task::spawn(self::task(self));
     }
 
-    async fn run_inner(&mut self) -> anyhow::Result<()> {
-        loop {
-            let Some(message) = self.receiver.recv().await else {
-                return Ok(());
-            };
-
-            log::trace!(r#"Processing message: "{message:?}""#);
-
-            let block_data = match self.find_block_data(message.block) {
-                Some(block_data) => block_data,
-                None => self.retreive_block_data(message.block).await?,
-            };
-
-            log::trace!(r#"Found data for the message block: "{block_data:?}""#);
-
-            self.process_message_block(message, block_data).await?;
-        }
-    }
-
     fn find_block_data(&self, block_number: u32) -> Option<BlockData> {
         self.blocks.find_by_block_number(block_number).cloned()
     }
@@ -86,56 +67,87 @@ impl MessageDataExtractor {
 
         Ok(block_data)
     }
-
-    async fn process_message_block(
-        &self,
-        message: Message,
-        block_data: BlockData,
-    ) -> anyhow::Result<()> {
-        let (block, authority_set_id) = block_data;
-        let messages = common::message_queued_events_of(&block);
-        let block_hash = block.hash();
-        for message_queued in messages {
-            if U256::from_big_endian(&message_queued.nonce_be).0 != message.nonce.0 {
-                log::info!("Message nonce mismatch, skipping {message_queued:?}");
-                continue;
-            }
-
-            log::trace!("Processing message in block: {message_queued:?}");
-
-            self.sender.send(MessageInBlock {
-                message: message_queued,
-                block: GearBlockNumber(block.number()),
-                block_hash: block_hash.0.into(),
-                authority_set_id,
-            })?;
-
-            break;
-        }
-
-        Ok(())
-    }
 }
 
 async fn task(mut this: MessageDataExtractor) {
+    let mut message_last = None;
     loop {
-        let result = this.run_inner().await;
+        let result = run_inner(&mut this, &mut message_last).await;
         let Err(e) = result else {
             log::trace!("Message data extractor exiting...");
             return;
         };
 
         log::error!("Message data extractor failed: {e}");
+        if let Err(e) = this.api_provider.reconnect().await {
+            log::error!(r#"Unable to reconnect: "{e}""#);
+            return;
+        }
 
-        match this.api_provider.reconnect().await {
-            Ok(_) => {
-                log::info!("Message queued extractor reconnected");
+        log::debug!("API provider reconnected");
+    }
+}
+
+async fn run_inner(
+    this: &mut MessageDataExtractor,
+    message_last: &mut Option<Message>,
+) -> anyhow::Result<()> {
+    loop {
+        let message = match message_last.clone() {
+            Some(message) => message,
+            None => {
+                let Some(message) = this.receiver.recv().await else {
+                    return Ok(());
+                };
+
+                *message_last = Some(message.clone());
+
+                message
             }
+        };
+
+        log::trace!(r#"Processing message: "{message:?}""#);
 
-            Err(e) => {
-                log::error!("Message queued extractor unable to reconnect: {e:?}");
-                return;
+        let block_data = match this.find_block_data(message.block) {
+            Some(block_data) => block_data,
+            None => this.retreive_block_data(message.block).await?,
+        };
+
+        *message_last = None;
+
+        log::trace!(r#"Found data for the message block: "{block_data:?}""#);
+
+        let (block, authority_set_id) = block_data;
+        let messages = common::message_queued_events_of(&block);
+        let block_hash = block.hash();
+        let Some(sender) = this.sender.upgrade() else {
+            log::info!("Unable to upgrade sender channel.");
+
+            return Ok(());
+        };
+
+        for message_queued in messages {
+            if U256::from_big_endian(&message_queued.nonce_be).0 != message.nonce.0 {
+                log::info!("Message nonce mismatch, skipping {message_queued:?}");
+                continue;
             }
+
+            log::trace!("Processing message in block: {message_queued:?}");
+
+            if sender
+                .send(MessageInBlock {
+                    message: message_queued,
+                    block: GearBlockNumber(block.number()),
+                    block_hash: block_hash.0.into(),
+                    authority_set_id,
+                })
+                .is_err()
+            {
+                log::info!("Sender channel closed.");
+                return Ok(());
+            }
+
+            break;
         }
     }
 }

relayer/src/message_relayer/common/gear/message_data_extractor.rs

@@ -13,6 +13,7 @@ pub struct PaidMessagesFilter {
     pending_messages: HashMap<[u8; 32], MessageInBlock>,
     pending_nonces: Vec<[u8; 32]>,
     excluded_from_fees: HashSet<AccountId32>,
+    sender: UnboundedSender<MessageInBlock>,
 
     metrics: Metrics,
 }
@@ -33,11 +34,15 @@ impl_metered_service! {
 }
 
 impl PaidMessagesFilter {
-    pub fn new(excluded_from_fees: HashSet<AccountId32>) -> Self {
+    pub fn new(
+        excluded_from_fees: HashSet<AccountId32>,
+        sender: UnboundedSender<MessageInBlock>,
+    ) -> Self {
         Self {
             pending_messages: HashMap::default(),
             pending_nonces: vec![],
             excluded_from_fees,
+            sender,
             metrics: Metrics::new(),
         }
     }
@@ -46,10 +51,9 @@ impl PaidMessagesFilter {
         mut self,
         mut messages: UnboundedReceiver<MessageInBlock>,
         mut paid_messages: UnboundedReceiver<PaidMessage>,
-        sender: UnboundedSender<MessageInBlock>,
     ) {
         tokio::spawn(async move {
-            match run_inner(&mut self, &sender, &mut messages, &mut paid_messages).await {
+            match run_inner(&mut self, &mut messages, &mut paid_messages).await {
                 Ok(_) => {}
                 Err(e) => log::error!("Paid messages filter failed: {e}"),
             }
@@ -59,7 +63,6 @@ impl PaidMessagesFilter {
 
 async fn run_inner(
     self_: &mut PaidMessagesFilter,
-    sender: &UnboundedSender<MessageInBlock>,
     messages: &mut UnboundedReceiver<MessageInBlock>,
     paid_messages: &mut UnboundedReceiver<PaidMessage>,
 ) -> anyhow::Result<()> {
@@ -91,7 +94,7 @@ async fn run_inner(
                         AccountId32::from(message.message.source),
                         hex::encode(message.message.nonce_be)
                     );
-                    sender.send(message)?;
+                    self_.sender.send(message)?;
 
                     continue;
                 }
@@ -100,7 +103,7 @@ async fn run_inner(
                     .pending_messages
                     .insert(message.message.nonce_be, message)
                 {
-                    panic!(
+                    log::error!(
                         "Received 2 messages with the same nonce: {}",
                         hex::encode(msg.message.nonce_be)
                     );
@@ -112,7 +115,7 @@ async fn run_inner(
 
         for i in (0..self_.pending_nonces.len()).rev() {
             if let Some(message) = self_.pending_messages.remove(&self_.pending_nonces[i]) {
-                sender.send(message)?;
+                self_.sender.send(message)?;
                 self_.pending_nonces.remove(i);
             }
         }
@@ -140,7 +143,8 @@ mod tests {
 
         set.insert(account0.into());
 
-        let filter = PaidMessagesFilter::new(set);
+        let (filter_msg_sender, mut msg_receiver) = mpsc::unbounded_channel();
+        let filter = PaidMessagesFilter::new(set, filter_msg_sender);
 
         let message0 = MessageInBlock {
             message: Message {
@@ -168,8 +172,7 @@ mod tests {
 
         let (msg_sender, filter_msg_receiver) = mpsc::unbounded_channel();
         let (paid_sender, paid_receiver) = mpsc::unbounded_channel();
-        let (filter_msg_sender, mut msg_receiver) = mpsc::unbounded_channel();
-        filter.spawn(filter_msg_receiver, paid_receiver, filter_msg_sender);
+        filter.spawn(filter_msg_receiver, paid_receiver);
 
         msg_sender.send(message0).unwrap();
         let res = msg_receiver.recv().await.unwrap();