XDM: Message submission checks to avoid validating duplicate XDM

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

@@ -32,6 +32,7 @@ use sp_domains_fraud_proof::storage_proof::FraudProofStorageKeyRequest;
 use sp_messenger::messages::{
     BlockMessagesWithStorageKey, ChainId, ChannelId, CrossDomainMessage, MessageId, MessageKey,
 };
+use sp_messenger::{ChannelNonce, XdmId};
 use sp_runtime::traits::{Block as BlockT, NumberFor};
 use sp_runtime::transaction_validity::{TransactionSource, TransactionValidity};
 use sp_runtime::{ApplyExtrinsicResult, ExtrinsicInclusionMode};
@@ -361,6 +362,14 @@ sp_api::impl_runtime_apis! {
         fn domain_chains_allowlist_update(_domain_id: DomainId) -> Option<DomainAllowlistUpdates>{
             unreachable!()
         }
+
+        fn xdm_id(_ext: &<Block as BlockT>::Extrinsic) -> Option<XdmId> {
+            unreachable!()
+        }
+
+        fn channel_nonce(_chain_id: ChainId, _channel_id: ChannelId) -> Option<ChannelNonce> {
+            unreachable!()
+        }
     }
 
     impl sp_messenger::RelayerApi<Block, BlockNumber, BlockNumber, <Block as BlockT>::Hash> for Runtime {

crates/subspace-malicious-operator/src/bin/subspace-malicious-operator.rs

@@ -355,7 +355,6 @@ fn main() -> Result<(), Error> {
                                         _,
                                         _,
                                         _,
-                                        DomainBlock,
                                         _,
                                         _,
                                     >(

crates/subspace-node/src/commands/run.rs

@@ -212,7 +212,6 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
                                 _,
                                 _,
                                 _,
-                                DomainBlock,
                                 _,
                                 _,
                             >(

crates/subspace-runtime/src/lib.rs

@@ -83,6 +83,7 @@ use sp_messenger::endpoint::{Endpoint, EndpointHandler as EndpointHandlerT, Endp
 use sp_messenger::messages::{
     BlockMessagesWithStorageKey, ChainId, CrossDomainMessage, FeeModel, MessageId, MessageKey,
 };
+use sp_messenger::{ChannelNonce, XdmId};
 use sp_messenger_host_functions::{get_storage_key, StorageKeyRequest};
 use sp_mmr_primitives::EncodableOpaqueLeaf;
 use sp_runtime::traits::{
@@ -1442,6 +1443,22 @@ impl_runtime_apis! {
         fn domain_chains_allowlist_update(domain_id: DomainId) -> Option<DomainAllowlistUpdates>{
             Messenger::domain_chains_allowlist_update(domain_id)
         }
+
+        fn xdm_id(ext: &<Block as BlockT>::Extrinsic) -> Option<XdmId> {
+            match &ext.function {
+                RuntimeCall::Messenger(pallet_messenger::Call::relay_message { msg })=> {
+                    Some(XdmId::RelayMessage((msg.src_chain_id, msg.channel_id, msg.nonce)))
+                }
+                RuntimeCall::Messenger(pallet_messenger::Call::relay_message_response { msg }) => {
+                    Some(XdmId::RelayResponseMessage((msg.src_chain_id, msg.channel_id, msg.nonce)))
+                }
+                _ => None,
+            }
+        }
+
+        fn channel_nonce(chain_id: ChainId, channel_id: ChannelId) -> Option<ChannelNonce> {
+            Messenger::channel_nonce(chain_id, channel_id)
+        }
     }
 
     impl sp_messenger::RelayerApi<Block, BlockNumber, BlockNumber, <Block as BlockT>::Hash> for Runtime {

domains/client/cross-domain-message-gossip/src/aux_schema.rs

@@ -1,10 +1,13 @@
 //! Schema for channel update storage.
 
+use crate::message_listener::LOG_TARGET;
 use parity_scale_codec::{Decode, Encode};
 use sc_client_api::backend::AuxStore;
-use sp_blockchain::{Error as ClientError, Result as ClientResult};
+use sp_blockchain::{Error as ClientError, Info, Result as ClientResult};
 use sp_core::H256;
 use sp_messenger::messages::{ChainId, ChannelId, ChannelState, Nonce};
+use sp_messenger::{ChannelNonce, XdmId};
+use sp_runtime::traits::{Block as BlockT, NumberFor};
 use subspace_runtime_primitives::BlockNumber;
 
 const CHANNEL_DETAIL: &[u8] = b"channel_detail";
@@ -86,3 +89,176 @@ where
         vec![],
     )
 }
+
+mod xdm_keys {
+    use parity_scale_codec::Encode;
+    use sp_domains::{ChainId, ChannelId};
+    use sp_messenger::messages::MessageKey;
+    use sp_messenger::XdmId;
+
+    const XDM: &[u8] = b"xdm";
+    const XDM_RELAY: &[u8] = b"relay_msg";
+    const XDM_RELAY_RESPONSE: &[u8] = b"relay_msg_response";
+    const XDM_LAST_CLEANUP_NONCE: &[u8] = b"xdm_last_cleanup_nonce";
+
+    pub(super) fn get_key_for_xdm_id(xdm_id: XdmId) -> Vec<u8> {
+        match xdm_id {
+            XdmId::RelayMessage(id) => get_key_for_xdm_relay(id),
+            XdmId::RelayResponseMessage(id) => get_key_for_xdm_relay_response(id),
+        }
+    }
+
+    pub(super) fn get_key_for_last_cleanup_relay_nonce(
+        chain_id: ChainId,
+        channel_id: ChannelId,
+    ) -> Vec<u8> {
+        (XDM, XDM_RELAY, XDM_LAST_CLEANUP_NONCE, chain_id, channel_id).encode()
+    }
+
+    pub(super) fn get_key_for_last_cleanup_relay_response_nonce(
+        chain_id: ChainId,
+        channel_id: ChannelId,
+    ) -> Vec<u8> {
+        (
+            XDM,
+            XDM_RELAY_RESPONSE,
+            XDM_LAST_CLEANUP_NONCE,
+            chain_id,
+            channel_id,
+        )
+            .encode()
+    }
+
+    pub(super) fn get_key_for_xdm_relay(id: MessageKey) -> Vec<u8> {
+        (XDM, XDM_RELAY, id).encode()
+    }
+
+    pub(super) fn get_key_for_xdm_relay_response(id: MessageKey) -> Vec<u8> {
+        (XDM, XDM_RELAY_RESPONSE, id).encode()
+    }
+}
+
+#[derive(Debug, Encode, Decode, Clone)]
+pub(super) struct BlockId<Block: BlockT> {
+    pub(super) number: NumberFor<Block>,
+    pub(super) hash: Block::Hash,
+}
+
+impl<Block: BlockT> From<Info<Block>> for BlockId<Block> {
+    fn from(value: Info<Block>) -> Self {
+        BlockId {
+            number: value.best_number,
+            hash: value.best_hash,
+        }
+    }
+}
+
+/// Store the given XDM ID as processed at given block.
+pub fn set_xdm_message_processed_at<Backend, Block>(
+    backend: &Backend,
+    xdm_id: XdmId,
+    block_id: BlockId<Block>,
+) -> ClientResult<()>
+where
+    Backend: AuxStore,
+    Block: BlockT,
+{
+    let key = xdm_keys::get_key_for_xdm_id(xdm_id);
+    backend.insert_aux(&[(key.as_slice(), block_id.encode().as_slice())], vec![])
+}
+
+/// Returns the maybe last processed block number for given xdm.
+pub fn get_xdm_processed_block_number<Backend, Block>(
+    backend: &Backend,
+    xdm_id: XdmId,
+) -> ClientResult<Option<BlockId<Block>>>
+where
+    Backend: AuxStore,
+    Block: BlockT,
+{
+    load_decode(backend, xdm_keys::get_key_for_xdm_id(xdm_id).as_slice())
+}
+
+/// Cleans up all the xdm storages until the latest nonces.
+pub fn cleanup_chain_channel_storages<Backend>(
+    backend: &Backend,
+    chain_id: ChainId,
+    channel_id: ChannelId,
+    channel_nonce: ChannelNonce,
+) -> ClientResult<()>
+where
+    Backend: AuxStore,
+{
+    let mut to_insert = vec![];
+    let mut to_delete = vec![];
+    if let Some(latest_relay_nonce) = channel_nonce.relay_msg_nonce {
+        let last_cleanup_relay_nonce_key =
+            xdm_keys::get_key_for_last_cleanup_relay_nonce(chain_id, channel_id);
+        let last_cleaned_up_nonce =
+            load_decode::<_, Nonce>(backend, last_cleanup_relay_nonce_key.as_slice())?;
+
+        let mut from_nonce = match last_cleaned_up_nonce {
+            None => Nonce::zero(),
+            Some(last_nonce) => last_nonce.saturating_add(Nonce::one()),
+        };
+
+        tracing::debug!(
+            target: LOG_TARGET,
+            "Cleaning Relay xdm keys for {:?} channel: {:?} from: {:?} to: {:?}",
+            chain_id,
+            channel_id,
+            from_nonce,
+            latest_relay_nonce
+        );
+
+        while from_nonce <= latest_relay_nonce {
+            to_delete.push(xdm_keys::get_key_for_xdm_relay((
+                chain_id, channel_id, from_nonce,
+            )));
+            from_nonce = from_nonce.saturating_add(Nonce::one());
+        }
+
+        to_insert.push((last_cleanup_relay_nonce_key, latest_relay_nonce.encode()));
+    }
+
+    if let Some(latest_relay_response_nonce) = channel_nonce.relay_response_msg_nonce {
+        let last_cleanup_relay_response_nonce_key =
+            xdm_keys::get_key_for_last_cleanup_relay_response_nonce(chain_id, channel_id);
+        let last_cleaned_up_nonce =
+            load_decode::<_, Nonce>(backend, last_cleanup_relay_response_nonce_key.as_slice())?;
+
+        let mut from_nonce = match last_cleaned_up_nonce {
+            None => Nonce::zero(),
+            Some(last_nonce) => last_nonce.saturating_add(Nonce::one()),
+        };
+
+        tracing::debug!(
+            target: LOG_TARGET,
+            "Cleaning Relay response xdm keys for {:?} channel: {:?} from: {:?} to: {:?}",
+            chain_id,
+            channel_id,
+            from_nonce,
+            latest_relay_response_nonce
+        );
+
+        while from_nonce <= latest_relay_response_nonce {
+            to_delete.push(xdm_keys::get_key_for_xdm_relay_response((
+                chain_id, channel_id, from_nonce,
+            )));
+            from_nonce = from_nonce.saturating_add(Nonce::one());
+        }
+
+        to_insert.push((
+            last_cleanup_relay_response_nonce_key,
+            latest_relay_response_nonce.encode(),
+        ));
+    }
+
+    backend.insert_aux(
+        &to_insert
+            .iter()
+            .map(|(k, v)| (k.as_slice(), v.as_slice()))
+            .collect::<Vec<_>>(),
+        &to_delete.iter().map(|k| k.as_slice()).collect::<Vec<_>>(),
+    )
+}

domains/client/cross-domain-message-gossip/src/gossip_worker.rs

@@ -314,4 +314,8 @@ pub(crate) mod rep {
     /// Reputation change when a peer sends us a gossip message that can't be decoded.
     pub(crate) const GOSSIP_NOT_DECODABLE: ReputationChange =
         ReputationChange::new_fatal("Cross chain message: not decodable");
+
+    /// Reputation change when a peer sends us a non XDM message
+    pub(crate) const NOT_XDM: ReputationChange =
+        ReputationChange::new_fatal("Cross chain message: not XDM");
 }

domains/client/cross-domain-message-gossip/src/lib.rs

@@ -1,3 +1,4 @@
+#![feature(let_chains)]
 #![warn(rust_2018_idioms)]
 
 mod aux_schema;