Remove abstract mutex and channel types from runtime context (#592)

* Remove HasMutex trait and use futures::Mutex directly

* Remove channel traits

* Fix use of futures::UnboundedReceiver, which works differently from Tokio's version

* Remove run_concurrent_task_stream method

* Remove stream traits

Author: soareschen

Cargo.lock

@@ -10,6 +10,8 @@ use cgp::core::error::{
 };
 use cgp::prelude::*;
 use eyre::Report;
+use futures::channel::mpsc::SendError;
+use futures::channel::oneshot::Canceled;
 use hermes_any_counterparty::impls::encoding::client_state::UnknownClientStateType;
 use hermes_any_counterparty::impls::encoding::consensus_state::UnknownConsensusStateType;
 use hermes_chain_type_components::traits::types::amount::HasAmountType;
@@ -132,6 +134,8 @@ delegate_components! {
             subtle_encoding::Error,
             reqwest::Error,
             InvalidUri,
+            SendError,
+            Canceled,
 
             // TODO: make it retryable?
             TransportError,

crates/cosmos/cosmos-relayer/src/impls/error.rs

@@ -17,3 +17,5 @@ hermes-relayer-components       = { workspace = true }
 hermes-chain-type-components    = { workspace = true }
 hermes-runtime-components       = { workspace = true }
 hermes-logging-components       = { workspace = true }
+
+futures                         = { workspace = true }

crates/relayer/relayer-components-extra/Cargo.toml

@@ -1,50 +1,49 @@
 use alloc::vec::Vec;
 use core::marker::PhantomData;
 
-use cgp::core::error::ErrorOf;
 use cgp::prelude::*;
+use futures::channel::mpsc::{SendError, TrySendError};
+use futures::channel::oneshot;
+use futures::channel::oneshot::Canceled;
 use hermes_chain_type_components::traits::types::message_response::MessageResponseOf;
 use hermes_relayer_components::chain::types::aliases::MessageOf;
 use hermes_relayer_components::relay::traits::ibc_message_sender::{
     CanSendIbcMessages, IbcMessageSender, IbcMessageSenderComponent,
 };
 use hermes_relayer_components::relay::traits::target::{HasTargetChainTypes, RelayTarget};
-use hermes_runtime_components::traits::channel::CanUseChannels;
-use hermes_runtime_components::traits::channel_once::{CanCreateChannelsOnce, CanUseChannelsOnce};
 
 use crate::batch::traits::channel::HasMessageBatchSender;
 use crate::batch::traits::types::CanUseMessageBatchChannel;
 use crate::batch::types::sink::BatchWorkerSink;
 
-pub struct SendMessagesToBatchWorker;
-
-#[cgp_provider(IbcMessageSenderComponent<Sink>)]
+#[cgp_new_provider(IbcMessageSenderComponent<Sink>)]
 impl<Relay, Sink, Target> IbcMessageSender<Relay, Sink, Target> for SendMessagesToBatchWorker
 where
     Target: RelayTarget,
     Relay: HasTargetChainTypes<Target>
         + CanSendIbcMessages<BatchWorkerSink, Target>
         + CanUseMessageBatchChannel<Target::Chain>
         + HasMessageBatchSender<Target::Chain>
-        + CanRaiseAsyncError<ErrorOf<Relay::Runtime>>,
-    Relay::Runtime: CanCreateChannelsOnce + CanUseChannelsOnce + CanUseChannels,
+        + CanRaiseAsyncError<SendError>
+        + CanRaiseAsyncError<Canceled>,
 {
     async fn send_messages(
         relay: &Relay,
         _target: Target,
         messages: Vec<MessageOf<Relay::TargetChain>>,
     ) -> Result<Vec<MessageResponseOf<Relay::TargetChain>>, Relay::Error> {
-        let (result_sender, result_receiver) = Relay::Runtime::new_channel_once();
+        let (result_sender, result_receiver) = oneshot::channel();
 
         let message_sender = relay.get_batch_sender(PhantomData::<Target::Chain>);
 
-        Relay::Runtime::send(message_sender, (messages, result_sender))
+        message_sender
+            .lock()
             .await
+            .unbounded_send((messages, result_sender))
+            .map_err(TrySendError::into_send_error)
             .map_err(Relay::raise_error)?;
 
-        let events = Relay::Runtime::receive_once(result_receiver)
-            .await
-            .map_err(Relay::raise_error)??;
+        let events = result_receiver.await.map_err(Relay::raise_error)??;
 
         Ok(events)
     }

crates/relayer/relayer-components-extra/src/batch/impls/message_sender.rs

@@ -1,16 +1,17 @@
+use alloc::sync::Arc;
 use alloc::vec::Vec;
 
 use cgp::core::Async;
 use cgp::prelude::HasAsyncErrorType;
+use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use futures::channel::oneshot;
+use futures::lock::Mutex;
 use hermes_chain_type_components::traits::types::message::HasMessageType;
 use hermes_chain_type_components::traits::types::message_response::{
     HasMessageResponseType, MessageResponseOf,
 };
 use hermes_relayer_components::chain::types::aliases::MessageOf;
 use hermes_relayer_components::multi::traits::chain_at::HasChainTypeAt;
-use hermes_runtime_components::traits::channel::{HasChannelTypes, ReceiverOf, SenderOf};
-use hermes_runtime_components::traits::channel_once::{HasChannelOnceTypes, SenderOnceOf};
-use hermes_runtime_components::traits::runtime::HasRuntimeType;
 
 pub trait HasMessageBatchChannelTypes<Tag>: Async {
     type BatchSubmission: Async;
@@ -26,62 +27,49 @@ pub type MessageBatchSenderOf<Context, Tag> =
 pub type MessageBatchReceiverOf<Context, Tag> =
     <Context as HasMessageBatchChannelTypes<Tag>>::MessageBatchReceiver;
 
-impl<Context, Tag, Chain, Runtime> HasMessageBatchChannelTypes<Tag> for Context
+impl<Context, Tag, Chain> HasMessageBatchChannelTypes<Tag> for Context
 where
-    Context:
-        HasChainTypeAt<Tag, Chain = Chain> + HasRuntimeType<Runtime = Runtime> + HasAsyncErrorType,
+    Context: HasChainTypeAt<Tag, Chain = Chain> + HasAsyncErrorType,
     Chain: HasMessageType + HasMessageResponseType,
-    Runtime: HasChannelTypes + HasChannelOnceTypes,
 {
     type BatchSubmission = (
         Vec<Chain::Message>,
-        SenderOnceOf<Runtime, Result<Vec<Chain::MessageResponse>, Context::Error>>,
+        oneshot::Sender<Result<Vec<Chain::MessageResponse>, Context::Error>>,
     );
 
-    type MessageBatchSender = SenderOf<Runtime, Self::BatchSubmission>;
+    type MessageBatchSender = Arc<Mutex<UnboundedSender<Self::BatchSubmission>>>;
 
-    type MessageBatchReceiver = ReceiverOf<Runtime, Self::BatchSubmission>;
+    type MessageBatchReceiver = UnboundedReceiver<Self::BatchSubmission>;
 }
 
 pub trait CanUseMessageBatchChannel<Tag>:
     HasChainTypeAt<Tag, Chain: HasMessageType + HasMessageResponseType>
-    + HasRuntimeType<Runtime: HasChannelTypes + HasChannelOnceTypes>
     + HasAsyncErrorType
     + HasMessageBatchChannelTypes<
         Tag,
         BatchSubmission = (
             Vec<MessageOf<Self::Chain>>,
-            SenderOnceOf<Self::Runtime, Result<Vec<MessageResponseOf<Self::Chain>>, Self::Error>>,
+            oneshot::Sender<Result<Vec<MessageResponseOf<Self::Chain>>, Self::Error>>,
         ),
-        MessageBatchSender = SenderOf<
-            Self::Runtime,
-            (
-                Vec<MessageOf<Self::Chain>>,
-                SenderOnceOf<
-                    Self::Runtime,
-                    Result<Vec<MessageResponseOf<Self::Chain>>, Self::Error>,
-                >,
-            ),
-        >,
-        MessageBatchReceiver = ReceiverOf<
-            Self::Runtime,
-            (
-                Vec<MessageOf<Self::Chain>>,
-                SenderOnceOf<
-                    Self::Runtime,
-                    Result<Vec<MessageResponseOf<Self::Chain>>, Self::Error>,
-                >,
-            ),
+        MessageBatchSender = Arc<
+            Mutex<
+                UnboundedSender<(
+                    Vec<MessageOf<Self::Chain>>,
+                    oneshot::Sender<Result<Vec<MessageResponseOf<Self::Chain>>, Self::Error>>,
+                )>,
+            >,
         >,
+        MessageBatchReceiver = UnboundedReceiver<(
+            Vec<MessageOf<Self::Chain>>,
+            oneshot::Sender<Result<Vec<MessageResponseOf<Self::Chain>>, Self::Error>>,
+        )>,
     >
 {
 }
 
-impl<Context, Tag, Chain, Runtime> CanUseMessageBatchChannel<Tag> for Context
+impl<Context, Tag, Chain> CanUseMessageBatchChannel<Tag> for Context
 where
-    Context:
-        HasChainTypeAt<Tag, Chain = Chain> + HasRuntimeType<Runtime = Runtime> + HasAsyncErrorType,
+    Context: HasChainTypeAt<Tag, Chain = Chain> + HasAsyncErrorType,
     Chain: HasMessageType + HasMessageResponseType,
-    Runtime: HasChannelTypes + HasChannelOnceTypes,
 {
 }

crates/relayer/relayer-components-extra/src/batch/traits/types.rs

@@ -13,9 +13,6 @@ use hermes_relayer_components::chain::traits::types::message::{
 use hermes_relayer_components::multi::traits::chain_at::HasChainAt;
 use hermes_relayer_components::relay::traits::ibc_message_sender::CanSendIbcMessages;
 use hermes_relayer_components::relay::traits::target::RelayTarget;
-use hermes_runtime_components::traits::channel::{CanUseChannels, HasChannelTypes};
-use hermes_runtime_components::traits::channel_once::{CanUseChannelsOnce, HasChannelOnceTypes};
-use hermes_runtime_components::traits::mutex::HasMutex;
 use hermes_runtime_components::traits::runtime::HasRuntime;
 use hermes_runtime_components::traits::sleep::CanSleep;
 use hermes_runtime_components::traits::spawn::CanSpawnTask;
@@ -100,7 +97,7 @@ where
         + CanUseMessageBatchChannel<Target::Chain>
         + CanProcessMessageBatches<Target>
         + for<'a> CanLog<LogBatchWorker<'a, Target>>,
-    Relay::Runtime: HasTime + HasMutex + CanSleep + CanUseChannels + HasChannelOnceTypes,
+    Relay::Runtime: HasTime + CanSleep,
 {
     async fn run_loop(&self, config: &BatchConfig, mut receiver: Relay::MessageBatchReceiver) {
         let runtime = self.runtime();
@@ -110,54 +107,56 @@ where
         let mut last_sent_time = runtime.now();
 
         loop {
-            let payload = Relay::Runtime::try_receive(&mut receiver);
-
-            match payload {
-                Ok(m_batch) => {
-                    if let Some(batch) = m_batch {
-                        let batch_size = batch.0.len();
-
-                        self.log(
-                            "received message batch",
-                            &LogBatchWorker {
-                                details: &format!("batch_size = {batch_size}"),
-                                log_level: LogLevel::Trace,
-                                phantom: PhantomData,
-                            },
-                        )
-                        .await;
-
-                        pending_batches.push_back(batch);
-                    }
+            // Poll and see if there is additional payload from the receiver.
+            // We do not block and wait for the next payload, as it may only arrive much later on.
+            // If there is no payload, we will still need to continue and process any pending messages
+            // that we have previously received.
+            let payload = receiver.try_next();
 
-                    let current_batch_size = pending_batches.len();
-                    let now = runtime.now();
+            // futures::UnboundedReceiver uses `Ok(None)` to indicate that the channel has closed.
+            // We ignore errors as it indicates the sender was not ready, which will be retried later on.
+            //
+            // https://docs.rs/futures/latest/futures/channel/mpsc/struct.UnboundedReceiver.html#method.try_next
+            let channel_closed = matches!(payload, Ok(None));
 
-                    self.process_message_batches(
-                        config,
-                        &mut pending_batches,
-                        now,
-                        &mut last_sent_time,
-                    )
-                    .await;
+            if let Ok(Some(batch)) = payload {
+                let batch_size = batch.0.len();
 
-                    if pending_batches.len() == current_batch_size {
-                        runtime.sleep(config.sleep_time).await;
-                    }
-                }
-                Err(e) => {
-                    self.log(
-                        "error in try_receive, terminating worker",
-                        &LogBatchWorker {
-                            details: &format!("error = {:?}", e),
-                            log_level: LogLevel::Error,
-                            phantom: PhantomData,
-                        },
-                    )
-                    .await;
-
-                    return;
-                }
+                self.log(
+                    "received message batch",
+                    &LogBatchWorker {
+                        details: &format!("batch_size = {batch_size}"),
+                        log_level: LogLevel::Trace,
+                        phantom: PhantomData,
+                    },
+                )
+                .await;
+
+                pending_batches.push_back(batch);
+            }
+
+            let current_batch_size = pending_batches.len();
+            let now = runtime.now();
+
+            self.process_message_batches(config, &mut pending_batches, now, &mut last_sent_time)
+                .await;
+
+            if pending_batches.len() == current_batch_size {
+                runtime.sleep(config.sleep_time).await;
+            }
+
+            if channel_closed {
+                self.log(
+                    "batch channel has closed, terminating worker",
+                    &LogBatchWorker {
+                        details: "",
+                        log_level: LogLevel::Info,
+                        phantom: PhantomData,
+                    },
+                )
+                .await;
+
+                return;
             }
         }
     }
@@ -184,8 +183,7 @@ where
         + CanUseMessageBatchChannel<Target::Chain>
         + CanPartitionMessageBatches<Target>
         + for<'a> CanLog<LogBatchWorker<'a, Target>>,
-    Relay::Runtime:
-        HasTime + CanSpawnTask + HasChannelTypes + HasChannelOnceTypes + HasAsyncErrorType,
+    Relay::Runtime: HasTime + CanSpawnTask + HasAsyncErrorType,
     SendReadyBatchTask<Relay, Target>: Task,
 {
     async fn process_message_batches(
@@ -325,7 +323,6 @@ where
     Relay: CanUseMessageBatchChannel<Target::Chain>
         + CanSendIbcMessages<BatchWorkerSink, Target>
         + for<'a> CanLog<LogBatchWorker<'a, Target>>,
-    Relay::Runtime: CanUseChannelsOnce + CanUseChannels,
     Relay::Error: Clone,
 {
     async fn send_ready_batches(&self, ready_batches: VecDeque<Relay::BatchSubmission>) {
@@ -366,7 +363,7 @@ where
                 .await;
 
                 for (_, sender) in senders.into_iter() {
-                    let _ = Relay::Runtime::send_once(sender, Err(e.clone()));
+                    let _ = sender.send(Err(e.clone()));
                 }
             }
             Ok(all_events) => {
@@ -385,7 +382,7 @@ where
 
                 for (message_count, sender) in senders.into_iter() {
                     let events = take(&mut all_events, message_count);
-                    let _ = Relay::Runtime::send_once(sender, Ok(events));
+                    let _ = sender.send(Ok(events));
                 }
             }
         }