Merge pull request #3384 from autonomys/xdm_relayer_limit

XDM: ensure relayer is only submitting XDMs that are within max nonce limit

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

@@ -396,6 +396,10 @@ sp_api::impl_runtime_apis! {
         fn channel_storage_key(_chain_id: ChainId, _channel_id: ChannelId) -> Vec<u8> {
             unreachable!()
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            unreachable!()
+        }
     }
 
     impl sp_domains_fraud_proof::FraudProofApi<Block, DomainHeader> for Runtime {

crates/subspace-runtime/src/lib.rs

@@ -1494,6 +1494,10 @@ impl_runtime_apis! {
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8> {
             Messenger::channel_storage_key(chain_id, channel_id)
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Messenger::open_channels()
+        }
     }
 
     impl sp_domains_fraud_proof::FraudProofApi<Block, DomainHeader> for Runtime {

domains/client/relayer/src/lib.rs

@@ -15,12 +15,12 @@ use parity_scale_codec::{Codec, Encode};
 use sc_client_api::{AuxStore, HeaderBackend, ProofProvider, StorageProof};
 use sc_utils::mpsc::TracingUnboundedSender;
 use sp_api::{ApiRef, ProvideRuntimeApi};
-use sp_core::H256;
+use sp_core::{H256, U256};
 use sp_domains::DomainsApi;
 use sp_messenger::messages::{
     BlockMessageWithStorageKey, BlockMessagesWithStorageKey, ChainId, CrossDomainMessage, Proof,
 };
-use sp_messenger::{MessengerApi, RelayerApi, XdmId};
+use sp_messenger::{MessengerApi, RelayerApi, XdmId, MAX_FUTURE_ALLOWED_NONCES};
 use sp_mmr_primitives::MmrApi;
 use sp_runtime::traits::{Block as BlockT, CheckedSub, Header as HeaderT, NumberFor, One};
 use sp_runtime::ArithmeticError;
@@ -311,16 +311,23 @@ where
             .ok()
             .flatten()
     {
-        // if this message should relay,
-        // check if the dst_chain inbox nonce is more than message nonce,
-        // if so, skip relaying since message is already executed on dst_chain
-        let relay_message = msg.nonce >= dst_channel_state.next_inbox_nonce;
+        // if the dst_chain inbox nonce is more than message nonce, skip relaying since message is
+        // already executed on dst_chain.
+        // if the message nonce is more than the max allowed nonce on dst_chain, skip relaying message.
+        let max_messages_nonce_allowed = dst_channel_state
+            .next_inbox_nonce
+            .saturating_add(MAX_FUTURE_ALLOWED_NONCES.into());
+        let relay_message = msg.nonce >= dst_channel_state.next_inbox_nonce
+            && msg.nonce <= max_messages_nonce_allowed;
         if !relay_message {
             log::debug!(
                 target: LOG_TARGET,
-                "Skipping message relay from {:?} to {:?}",
+                "Skipping Outbox message relay from {:?} to {:?} of XDM[{:?}, {:?}]. Max Nonce allowed: {:?}",
                 msg.src_chain_id,
                 msg.dst_chain_id,
+                msg.channel_id,
+                msg.nonce,
+                max_messages_nonce_allowed
             );
             return false;
         }
@@ -363,17 +370,30 @@ where
         get_channel_state(backend, msg.dst_chain_id, msg.src_chain_id, msg.channel_id)
             .ok()
             .flatten()
-        && let Some(dst_chain_outbox_response_nonce) =
-            dst_channel_state.latest_response_received_message_nonce
     {
+        let next_nonce = if let Some(dst_chain_outbox_response_nonce) =
+            dst_channel_state.latest_response_received_message_nonce
+        {
+            // next nonce for outbox response will be +1 of current nonce for which
+            // dst_chain has received response.
+            dst_chain_outbox_response_nonce.saturating_add(U256::one())
+        } else {
+            // if dst_chain has not received the first outbox response,
+            // then next nonce will be just 0
+            U256::zero()
+        };
         // relay inbox response if the dst_chain did not execute is already
-        let relay_message = msg.nonce > dst_chain_outbox_response_nonce;
+        let max_msg_nonce_allowed = next_nonce.saturating_add(MAX_FUTURE_ALLOWED_NONCES.into());
+        let relay_message = msg.nonce >= next_nonce && msg.nonce <= max_msg_nonce_allowed;
         if !relay_message {
             log::debug!(
                 target: LOG_TARGET,
-                "Skipping message relay from {:?} to {:?}",
+                "Skipping Inbox response message relay from {:?} to {:?} of XDM[{:?}, {:?}]. Max nonce allowed: {:?}",
                 msg.src_chain_id,
                 msg.dst_chain_id,
+                msg.channel_id,
+                msg.nonce,
+                max_msg_nonce_allowed
             );
             return false;
         }

domains/client/relayer/src/worker.rs

@@ -2,13 +2,13 @@ use crate::{BlockT, Error, GossipMessageSink, HeaderBackend, HeaderT, Relayer, L
 use cross_domain_message_gossip::{ChannelUpdate, Message as GossipMessage, MessageData};
 use futures::StreamExt;
 use sc_client_api::{AuxStore, BlockchainEvents, ProofProvider};
-use sp_api::ProvideRuntimeApi;
+use sp_api::{ApiExt, ProvideRuntimeApi};
 use sp_consensus::SyncOracle;
 use sp_domains::{DomainId, DomainsApi};
 use sp_messenger::messages::ChainId;
 use sp_messenger::{MessengerApi, RelayerApi};
 use sp_mmr_primitives::MmrApi;
-use sp_runtime::traits::{CheckedSub, NumberFor, One};
+use sp_runtime::traits::{CheckedSub, NumberFor, One, Zero};
 use sp_runtime::SaturatedConversion;
 use std::sync::Arc;
 
@@ -103,9 +103,27 @@ where
 {
     let api = client.runtime_api();
 
-    let updated_channels = api.updated_channels(block_hash)?;
+    let channels_status_to_broadcast = {
+        let updated_channels = api.updated_channels(block_hash)?;
+
+        // TODO: remove version check before next network
+        let relayer_api_version = api
+            .api_version::<dyn RelayerApi<Block, NumberFor<Block>, NumberFor<CBlock>, CBlock::Hash>>(block_hash)?
+            // It is safe to return a default version of 1, since there will always be version 1.
+            .unwrap_or(1);
+
+        // if there are no channel updates, broadcast channel's status for every 300 blocks
+        if updated_channels.is_empty()
+            && relayer_api_version >= 2
+            && block_number % 300u32.into() == Zero::zero()
+        {
+            api.open_channels(block_hash)?
+        } else {
+            updated_channels
+        }
+    };
 
-    for (dst_chain_id, channel_id) in updated_channels {
+    for (dst_chain_id, channel_id) in channels_status_to_broadcast {
         let storage_key = api.channel_storage_key(block_hash, dst_chain_id, channel_id)?;
         let proof = client
             .read_proof(block_hash, &mut [storage_key.as_ref()].into_iter())

domains/pallets/messenger/src/lib.rs

@@ -48,16 +48,6 @@ use sp_messenger::messages::{
 use sp_runtime::traits::{Extrinsic, Hash};
 use sp_runtime::DispatchError;
 
-/// Maximum number of XDMs per domain/channel with future nonces that are allowed to be validated.
-/// Any XDM comes with a nonce above Maximum future nonce will be rejected.
-// TODO: We need to benchmark how many XDMs can fit in to a
-//  - Single consensus block
-//  - Single domain block(includes all bundles filled with XDMs)
-//  Once we have that info, we can make a better judgement on how many XDMs
-//  we want to include per block while allowing other extrinsics to be included as well.
-//  Note: Currently XDM takes priority over other extrinsics unless they come with priority fee
-const MAX_FUTURE_ALLOWED_NONCES: u32 = 256;
-
 /// Transaction validity for a given validated XDM extrinsic.
 /// If the extrinsic is not included in the bundle, extrinsic is removed from the TxPool.
 const XDM_TRANSACTION_LONGEVITY: u64 = 10;
@@ -133,7 +123,7 @@ mod pallet {
     use crate::{
         BalanceOf, ChainAllowlistUpdate, Channel, ChannelId, ChannelState, CloseChannelBy,
         FeeModel, HoldIdentifier, Nonce, OutboxMessageResult, StateRootOf, ValidatedRelayMessage,
-        MAX_FUTURE_ALLOWED_NONCES, STORAGE_VERSION, U256, XDM_TRANSACTION_LONGEVITY,
+        STORAGE_VERSION, U256, XDM_TRANSACTION_LONGEVITY,
     };
     #[cfg(not(feature = "std"))]
     use alloc::boxed::Box;
@@ -158,7 +148,7 @@ mod pallet {
     };
     use sp_messenger::{
         ChannelNonce, DomainRegistration, InherentError, InherentType, OnXDMRewards, StorageKeys,
-        INHERENT_IDENTIFIER,
+        INHERENT_IDENTIFIER, MAX_FUTURE_ALLOWED_NONCES,
     };
     use sp_runtime::traits::Zero;
     use sp_runtime::{ArithmeticError, Perbill, Saturating};
@@ -1399,6 +1389,19 @@ mod pallet {
             UpdatedChannels::<T>::get()
         }
 
+        pub fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Channels::<T>::iter().fold(
+                BTreeSet::new(),
+                |mut acc, (dst_chain_id, channel_id, channel)| {
+                    if channel.state != ChannelState::Closed {
+                        acc.insert((dst_chain_id, channel_id));
+                    }
+
+                    acc
+                },
+            )
+        }
+
         pub fn channel_nonce(chain_id: ChainId, channel_id: ChannelId) -> Option<ChannelNonce> {
             Channels::<T>::get(chain_id, channel_id).map(|channel| {
                 let last_inbox_nonce = channel.next_inbox_nonce.checked_sub(U256::one());

domains/primitives/messenger/src/lib.rs

@@ -42,6 +42,16 @@ use std::collections::BTreeSet;
 /// Messenger inherent identifier.
 pub const INHERENT_IDENTIFIER: InherentIdentifier = *b"messengr";
 
+/// Maximum number of XDMs per domain/channel with future nonces that are allowed to be validated.
+/// Any XDM comes with a nonce above Maximum future nonce will be rejected.
+// TODO: We need to benchmark how many XDMs can fit in to a
+//  - Single consensus block
+//  - Single domain block(includes all bundles filled with XDMs)
+//  Once we have that info, we can make a better judgement on how many XDMs
+//  we want to include per block while allowing other extrinsics to be included as well.
+//  Note: Currently XDM takes priority over other extrinsics unless they come with priority fee
+pub const MAX_FUTURE_ALLOWED_NONCES: u32 = 256;
+
 /// Trait to handle XDM rewards.
 pub trait OnXDMRewards<Balance> {
     fn on_xdm_rewards(rewards: Balance);
@@ -188,6 +198,7 @@ pub struct ChannelNonce {
 
 sp_api::decl_runtime_apis! {
     /// Api useful for relayers to fetch messages and submit transactions.
+    #[api_version(2)]
     pub trait RelayerApi<BlockNumber, CNumber, CHash>
     where
         BlockNumber: Encode + Decode,
@@ -219,6 +230,9 @@ sp_api::decl_runtime_apis! {
 
         /// Returns storage key for channels for given chain and channel id.
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8>;
+
+        /// Returns all the open channels to other chains.
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)>;
     }
 
     /// Api to provide XDM extraction from Runtime Calls.

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

@@ -960,6 +960,10 @@ impl_runtime_apis! {
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8> {
             Messenger::channel_storage_key(chain_id, channel_id)
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Messenger::open_channels()
+        }
     }
 
     impl sp_domain_sudo::DomainSudoApi<Block> for Runtime {

domains/runtime/evm/src/lib.rs

@@ -1418,6 +1418,10 @@ impl_runtime_apis! {
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8> {
             Messenger::channel_storage_key(chain_id, channel_id)
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Messenger::open_channels()
+        }
     }
 
     impl fp_rpc::EthereumRuntimeRPCApi<Block> for Runtime {

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

@@ -954,6 +954,10 @@ impl_runtime_apis! {
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8> {
             Messenger::channel_storage_key(chain_id, channel_id)
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Messenger::open_channels()
+        }
     }
 
     impl sp_domain_sudo::DomainSudoApi<Block> for Runtime {

domains/test/runtime/evm/src/lib.rs

@@ -1440,6 +1440,10 @@ impl_runtime_apis! {
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8> {
             Messenger::channel_storage_key(chain_id, channel_id)
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Messenger::open_channels()
+        }
     }
 
     impl fp_rpc::EthereumRuntimeRPCApi<Block> for Runtime {

test/subspace-test-runtime/src/lib.rs

@@ -1547,6 +1547,10 @@ impl_runtime_apis! {
         fn channel_storage_key(chain_id: ChainId, channel_id: ChannelId) -> Vec<u8> {
             Messenger::channel_storage_key(chain_id, channel_id)
         }
+
+        fn open_channels() -> BTreeSet<(ChainId, ChannelId)> {
+            Messenger::open_channels()
+        }
     }
 
     impl sp_domains_fraud_proof::FraudProofApi<Block, DomainHeader> for Runtime {