fix skipping channels that are closed and no further messages needs to be processed

Author: vedhavyas

crates/subspace-fake-runtime-api/src/lib.rs

@@ -16,7 +16,7 @@ use sp_domains::{
 use sp_domains_fraud_proof::fraud_proof::FraudProof;
 use sp_domains_fraud_proof::storage_proof::FraudProofStorageKeyRequest;
 use sp_messenger::messages::{
-    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelId, ChannelState,
+    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelId, ChannelStateWithNonce,
     CrossDomainMessage, MessageId, MessageKey, Nonce as XdmNonce,
 };
 use sp_messenger::{ChannelNonce, XdmId};
@@ -418,7 +418,7 @@ sp_api::impl_runtime_apis! {
             unreachable!()
         }
 
-        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelState)> {
+        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
             unreachable!()
         }
 

crates/subspace-runtime/src/lib.rs

@@ -71,8 +71,8 @@ use sp_domains_fraud_proof::storage_proof::{
 };
 use sp_messenger::endpoint::{Endpoint, EndpointHandler as EndpointHandlerT, EndpointId};
 use sp_messenger::messages::{
-    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelState, CrossDomainMessage,
-    MessageId, MessageKey, Nonce as XdmNonce,
+    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelStateWithNonce,
+    CrossDomainMessage, MessageId, MessageKey, Nonce as XdmNonce,
 };
 use sp_messenger::{ChannelNonce, XdmId};
 use sp_messenger_host_functions::{get_storage_key, StorageKeyRequest};
@@ -1708,7 +1708,7 @@ impl_runtime_apis! {
             Messenger::get_block_messages(query)
         }
 
-        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelState)> {
+        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
             Messenger::channels_and_states()
         }
 

domains/client/relayer/src/lib.rs

@@ -12,7 +12,7 @@ use crate::aux_schema::{
 };
 use async_channel::TrySendError;
 use cross_domain_message_gossip::{
-    get_channel_state, Message as GossipMessage, MessageData as GossipMessageData,
+    get_channel_state, ChannelDetail, Message as GossipMessage, MessageData as GossipMessageData,
 };
 use parity_scale_codec::{Codec, Encode};
 use rand::seq::SliceRandom;
@@ -23,7 +23,7 @@ use sp_core::H256;
 use sp_domains::{ChannelId, DomainsApi};
 use sp_messenger::messages::{
     BlockMessageWithStorageKey, BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId,
-    ChannelState, CrossDomainMessage, Nonce, Proof,
+    ChannelState, ChannelStateWithNonce, CrossDomainMessage, Nonce, Proof,
 };
 use sp_messenger::{MessengerApi, RelayerApi};
 use sp_mmr_primitives::MmrApi;
@@ -633,7 +633,7 @@ fn get_channel_state_query<Backend, Client, Block>(
     self_chain_id: ChainId,
     dst_chain_id: ChainId,
     channel_id: ChannelId,
-    local_channel_state: ChannelState,
+    local_channel_state: ChannelStateWithNonce,
 ) -> Result<Option<BlockMessagesQuery>, Error>
 where
     Backend: AuxStore,
@@ -667,16 +667,19 @@ where
             inbox_responses_from: Nonce::zero(),
         }),
         // don't have channel processed state, so use the dst_channel state for query
-        (Some(dst_channel_state), None, None) => Some(BlockMessagesQuery {
-            chain_id: dst_chain_id,
-            channel_id,
-            outbox_from: dst_channel_state.next_inbox_nonce,
-            inbox_responses_from: dst_channel_state
-                .latest_response_received_message_nonce
-                // pick the next inbox message response nonce or default to zero
-                .map(|nonce| nonce.saturating_add(One::one()))
-                .unwrap_or(Nonce::zero()),
-        }),
+        (Some(dst_channel_state), None, None) => {
+            should_relay_messages_to_channel(dst_chain_id, &dst_channel_state, local_channel_state)
+                .then_some(BlockMessagesQuery {
+                    chain_id: dst_chain_id,
+                    channel_id,
+                    outbox_from: dst_channel_state.next_inbox_nonce,
+                    inbox_responses_from: dst_channel_state
+                        .latest_response_received_message_nonce
+                        // pick the next inbox message response nonce or default to zero
+                        .map(|nonce| nonce.saturating_add(One::one()))
+                        .unwrap_or(Nonce::zero()),
+                })
+        }
         // don't have dst channel state, so use the last processed channel state
         (None, last_outbox_nonce, last_inbox_message_response_nonce) => Some(BlockMessagesQuery {
             chain_id: dst_chain_id,
@@ -689,61 +692,32 @@ where
                 .unwrap_or(Nonce::zero()),
         }),
         (Some(dst_channel_state), last_outbox_nonce, last_inbox_message_response_nonce) => {
-            let next_outbox_nonce = max(
-                dst_channel_state.next_inbox_nonce,
-                last_outbox_nonce
-                    .map(|nonce| nonce.saturating_add(One::one()))
-                    .unwrap_or(Nonce::zero()),
-            );
-
-            let next_inbox_response_nonce = max(
-                dst_channel_state
-                    .latest_response_received_message_nonce
-                    .map(|nonce| nonce.saturating_add(One::one()))
-                    .unwrap_or(Nonce::zero()),
-                last_inbox_message_response_nonce
-                    .map(|nonce| nonce.saturating_add(One::one()))
-                    .unwrap_or(Nonce::zero()),
-            );
-
-            // if the local channel is closed, and
-            // last outbox message is already included
-            // and
-            // if the dst_channel is closed, and
-            // last inbox message response is already included
-            // we can safely skip the channel as the there is nothing further to send.
-            if local_channel_state == ChannelState::Closed
-                && dst_channel_state.state == ChannelState::Closed
-            {
-                let is_last_outbox_nonce = dst_channel_state
-                    .next_inbox_nonce
-                    .saturating_sub(One::one())
-                    == last_outbox_nonce.unwrap_or(Nonce::zero());
-
-                let is_last_inbox_message_response_nonce = dst_channel_state
-                    .latest_response_received_message_nonce
-                    .unwrap_or(Nonce::zero())
-                    == last_inbox_message_response_nonce.unwrap_or(Nonce::zero());
-
-                if is_last_outbox_nonce && is_last_inbox_message_response_nonce {
-                    tracing::debug!(target: LOG_TARGET, "Skipping XDM for Chain[{:?}] - Channel[{:?}]", dst_chain_id, channel_id);
-                    None
-                } else {
-                    Some(BlockMessagesQuery {
+            should_relay_messages_to_channel(dst_chain_id, &dst_channel_state, local_channel_state)
+                .then(|| {
+                    let next_outbox_nonce = max(
+                        dst_channel_state.next_inbox_nonce,
+                        last_outbox_nonce
+                            .map(|nonce| nonce.saturating_add(One::one()))
+                            .unwrap_or(Nonce::zero()),
+                    );
+
+                    let next_inbox_response_nonce = max(
+                        dst_channel_state
+                            .latest_response_received_message_nonce
+                            .map(|nonce| nonce.saturating_add(One::one()))
+                            .unwrap_or(Nonce::zero()),
+                        last_inbox_message_response_nonce
+                            .map(|nonce| nonce.saturating_add(One::one()))
+                            .unwrap_or(Nonce::zero()),
+                    );
+
+                    BlockMessagesQuery {
                         chain_id: dst_chain_id,
                         channel_id,
                         outbox_from: next_outbox_nonce,
                         inbox_responses_from: next_inbox_response_nonce,
-                    })
-                }
-            } else {
-                Some(BlockMessagesQuery {
-                    chain_id: dst_chain_id,
-                    channel_id,
-                    outbox_from: next_outbox_nonce,
-                    inbox_responses_from: next_inbox_response_nonce,
+                    }
                 })
-            }
         }
     };
 
@@ -756,6 +730,45 @@ where
     Ok(query)
 }
 
+fn should_relay_messages_to_channel(
+    dst_chain_id: ChainId,
+    dst_channel_state: &ChannelDetail,
+    local_channel_state: ChannelStateWithNonce,
+) -> bool {
+    let should_process = if dst_channel_state.state == ChannelState::Closed
+        && let ChannelStateWithNonce::Closed {
+            next_outbox_nonce,
+            next_inbox_nonce,
+        } = local_channel_state
+    {
+        // if the next outbox nonce of local channel is same as
+        // next inbox nonce of dst_channel, then there are no further
+        // outbox messages to be sent from the local channel
+        let no_outbox_messages = next_outbox_nonce == dst_channel_state.next_inbox_nonce;
+
+        // if next inbox nonce of local channel is +1 of
+        // last received response nonce on dst_chain, then there are
+        // no further messages responses to be sent from the local channel
+        let no_inbox_responses_messages = dst_channel_state
+            .latest_response_received_message_nonce
+            .map(|nonce| nonce == next_inbox_nonce.saturating_sub(Nonce::one()))
+            .unwrap_or(false);
+
+        !(no_outbox_messages && no_inbox_responses_messages)
+    } else {
+        true
+    };
+
+    if !should_process {
+        tracing::debug!(target: LOG_TARGET,
+            "Chain[{:?}] for Channel[{:?}] is closed and no messages to process",
+            dst_chain_id, dst_channel_state.channel_id
+        );
+    }
+
+    should_process
+}
+
 fn is_relayer_api_version_available<Client, Block, CBlock>(
     client: &Arc<Client>,
     version: u32,

domains/pallets/messenger/src/lib.rs

@@ -174,8 +174,9 @@ mod pallet {
         Endpoint, EndpointHandler, EndpointRequest, EndpointRequestWithCollectedFee, Sender,
     };
     use sp_messenger::messages::{
-        ChainId, ChannelOpenParamsV1, CrossDomainMessage, Message, MessageId, MessageKey,
-        MessageWeightTag, PayloadV1, ProtocolMessageRequest, RequestResponse, VersionedPayload,
+        ChainId, ChannelOpenParamsV1, ChannelStateWithNonce, CrossDomainMessage, Message,
+        MessageId, MessageKey, MessageWeightTag, PayloadV1, ProtocolMessageRequest,
+        RequestResponse, VersionedPayload,
     };
     use sp_messenger::{
         ChannelNonce, DomainRegistration, InherentError, InherentType, NoteChainTransfer,
@@ -1414,7 +1415,7 @@ mod pallet {
             Channels::<T>::iter_keys().collect()
         }
 
-        pub fn channels_and_states() -> Vec<(ChainId, ChannelId, ChannelState)> {
+        pub fn channels_and_states() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
             crate::migrations::get_channels_and_states::<T>()
         }
 

domains/pallets/messenger/src/migrations/v1_to_v2.rs

@@ -77,6 +77,7 @@ impl<Balance, AccountId> From<Channel<Balance, AccountId>> for ChannelV1<Balance
 
 pub(crate) mod migrate_channels {
     use super::*;
+    use sp_messenger::messages::ChannelStateWithNonce;
     use sp_runtime::traits::Get;
 
     #[storage_alias]
@@ -110,12 +111,24 @@ pub(crate) mod migrate_channels {
         })
     }
 
-    pub(crate) fn get_channels_and_states<T: Config>() -> Vec<(ChainId, ChannelId, ChannelState)> {
+    pub(crate) fn get_channels_and_states<T: Config>(
+    ) -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
         let keys: Vec<(ChainId, ChannelId)> = ChannelStorageV1::<T>::iter_keys().collect();
         keys.into_iter()
             .filter_map(|(chain_id, channel_id)| {
-                get_channel::<T>(chain_id, channel_id)
-                    .map(|channel| (chain_id, channel_id, channel.state))
+                get_channel::<T>(chain_id, channel_id).map(|channel| {
+                    let state = channel.state;
+                    let state_with_nonce = match state {
+                        ChannelState::Initiated => ChannelStateWithNonce::Initiated,
+                        ChannelState::Open => ChannelStateWithNonce::Open,
+                        ChannelState::Closed => ChannelStateWithNonce::Closed {
+                            next_outbox_nonce: channel.next_outbox_nonce,
+                            next_inbox_nonce: channel.next_inbox_nonce,
+                        },
+                    };
+
+                    (chain_id, channel_id, state_with_nonce)
+                })
             })
             .collect()
     }

domains/primitives/messenger/src/lib.rs

@@ -23,7 +23,7 @@ pub mod messages;
 #[cfg(not(feature = "std"))]
 extern crate alloc;
 
-use crate::messages::{ChannelState, MessageKey, Nonce};
+use crate::messages::{ChannelStateWithNonce, MessageKey, Nonce};
 #[cfg(not(feature = "std"))]
 use alloc::collections::BTreeMap;
 #[cfg(not(feature = "std"))]
@@ -309,7 +309,7 @@ sp_api::decl_runtime_apis! {
         fn block_messages_with_query(query: BlockMessagesQuery) -> BlockMessagesWithStorageKey;
 
         /// Returns all the channels to other chains and their local Channel state.
-        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelState)>;
+        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)>;
 
         /// Returns the first outbox message nonce that should be relayed to the dst_chain.
         fn should_relay_outbox_messages(dst_chain_id: ChainId, channel_id: ChannelId, from_nonce: Nonce) -> Option<Nonce>;

domains/primitives/messenger/src/messages.rs

@@ -34,6 +34,21 @@ pub enum ChannelState {
     Closed,
 }
 
+/// State of channel and nonces when channel is closed.
+#[derive(Default, Debug, Encode, Decode, Clone, Eq, PartialEq, TypeInfo)]
+pub enum ChannelStateWithNonce {
+    /// Channel between chains is initiated but do not yet send or receive messages in this state.
+    #[default]
+    Initiated,
+    /// Channel is open and can send and receive messages.
+    Open,
+    /// Channel is closed along with nonces at current point.
+    Closed {
+        next_outbox_nonce: Nonce,
+        next_inbox_nonce: Nonce,
+    },
+}
+
 /// Channel describes a bridge to exchange messages between two chains.
 #[derive(Default, Debug, Encode, Decode, Clone, Eq, PartialEq, TypeInfo)]
 pub struct Channel<Balance, AccountId> {

domains/runtime/auto-id/src/lib.rs

@@ -42,8 +42,8 @@ use sp_core::{Get, OpaqueMetadata};
 use sp_domains::{ChannelId, DomainAllowlistUpdates, DomainId, Transfers};
 use sp_messenger::endpoint::{Endpoint, EndpointHandler as EndpointHandlerT, EndpointId};
 use sp_messenger::messages::{
-    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelState, CrossDomainMessage,
-    MessageId, MessageKey, Nonce as XdmNonce,
+    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelStateWithNonce,
+    CrossDomainMessage, MessageId, MessageKey, Nonce as XdmNonce,
 };
 use sp_messenger::{ChannelNonce, XdmId};
 use sp_messenger_host_functions::{get_storage_key, StorageKeyRequest};
@@ -1150,7 +1150,7 @@ impl_runtime_apis! {
             Messenger::get_block_messages(query)
         }
 
-        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelState)> {
+        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
             Messenger::channels_and_states()
         }
 

domains/runtime/evm/src/lib.rs

@@ -64,8 +64,8 @@ use sp_evm_tracker::{
 };
 use sp_messenger::endpoint::{Endpoint, EndpointHandler as EndpointHandlerT, EndpointId};
 use sp_messenger::messages::{
-    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelState, CrossDomainMessage,
-    MessageId, MessageKey, Nonce as XdmNonce,
+    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelStateWithNonce,
+    CrossDomainMessage, MessageId, MessageKey, Nonce as XdmNonce,
 };
 use sp_messenger::{ChannelNonce, XdmId};
 use sp_messenger_host_functions::{get_storage_key, StorageKeyRequest};
@@ -1579,7 +1579,7 @@ impl_runtime_apis! {
             Messenger::get_block_messages(query)
         }
 
-        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelState)> {
+        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
             Messenger::channels_and_states()
         }
 

domains/test/runtime/auto-id/src/lib.rs

@@ -42,8 +42,8 @@ use sp_core::{Get, OpaqueMetadata};
 use sp_domains::{ChannelId, DomainAllowlistUpdates, DomainId, Transfers};
 use sp_messenger::endpoint::{Endpoint, EndpointHandler as EndpointHandlerT, EndpointId};
 use sp_messenger::messages::{
-    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelState, CrossDomainMessage,
-    MessageId, MessageKey, Nonce as XdmNonce,
+    BlockMessagesQuery, BlockMessagesWithStorageKey, ChainId, ChannelStateWithNonce,
+    CrossDomainMessage, MessageId, MessageKey, Nonce as XdmNonce,
 };
 use sp_messenger::{ChannelNonce, XdmId};
 use sp_messenger_host_functions::{get_storage_key, StorageKeyRequest};
@@ -1146,7 +1146,7 @@ impl_runtime_apis! {
             Messenger::get_block_messages(query)
         }
 
-        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelState)> {
+        fn channels_and_state() -> Vec<(ChainId, ChannelId, ChannelStateWithNonce)> {
             Messenger::channels_and_states()
         }