Merge pull request #3310 from autonomys/xdm_relayer

vedhavyas · web-flow · commit aaed3baf5b13 · 2024-12-12T02:35:08.000Z
XDM: Cache and submit XDM messages from the relayer instead of submitting them every block until confirmation
diff --git a/domains/client/cross-domain-message-gossip/src/aux_schema.rs b/domains/client/cross-domain-message-gossip/src/aux_schema.rs
@@ -1,16 +1,18 @@
 //! Schema for channel update storage.
 
-use crate::message_listener::LOG_TARGET;
+use crate::RELAYER_PREFIX;
 use parity_scale_codec::{Decode, Encode};
 use sc_client_api::backend::AuxStore;
 use sp_blockchain::{Error as ClientError, Info, Result as ClientResult};
+use sp_core::bytes::to_hex;
 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";
+const LOG_TARGET: &str = "gossip_aux_schema";
 
 fn channel_detail_key(
     src_chain_id: ChainId,
@@ -58,35 +60,48 @@ pub struct ChannelDetail {
 /// Load the channel state of self_chain_id on chain_id.
 pub fn get_channel_state<Backend>(
     backend: &Backend,
-    self_chain_id: ChainId,
-    chain_id: ChainId,
+    dst_chain_id: ChainId,
+    src_chain_id: ChainId,
     channel_id: ChannelId,
 ) -> ClientResult<Option<ChannelDetail>>
 where
     Backend: AuxStore,
 {
     load_decode(
         backend,
-        channel_detail_key(chain_id, self_chain_id, channel_id).as_slice(),
+        channel_detail_key(src_chain_id, dst_chain_id, channel_id).as_slice(),
     )
 }
 
 /// Set the channel state of self_chain_id on chain_id.
 pub fn set_channel_state<Backend>(
     backend: &Backend,
-    self_chain_id: ChainId,
-    chain_id: ChainId,
+    dst_chain_id: ChainId,
+    src_chain_id: ChainId,
     channel_detail: ChannelDetail,
 ) -> ClientResult<()>
 where
     Backend: AuxStore,
 {
     backend.insert_aux(
         &[(
-            channel_detail_key(chain_id, self_chain_id, channel_detail.channel_id).as_slice(),
+            channel_detail_key(src_chain_id, dst_chain_id, channel_detail.channel_id).as_slice(),
             channel_detail.encode().as_slice(),
         )],
         vec![],
+    )?;
+
+    let channel_nonce = ChannelNonce {
+        relay_msg_nonce: Some(channel_detail.next_inbox_nonce),
+        relay_response_msg_nonce: channel_detail.latest_response_received_message_nonce,
+    };
+    let prefix = (RELAYER_PREFIX, src_chain_id).encode();
+    cleanup_chain_channel_storages(
+        backend,
+        &prefix,
+        src_chain_id,
+        channel_detail.channel_id,
+        channel_nonce,
     )
 }
 
@@ -101,25 +116,36 @@ mod xdm_keys {
     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> {
+    pub(super) fn get_key_for_xdm_id(prefix: &[u8], 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),
+            XdmId::RelayMessage(id) => get_key_for_xdm_relay(prefix, id),
+            XdmId::RelayResponseMessage(id) => get_key_for_xdm_relay_response(prefix, id),
         }
     }
 
     pub(super) fn get_key_for_last_cleanup_relay_nonce(
+        prefix: &[u8],
         chain_id: ChainId,
         channel_id: ChannelId,
     ) -> Vec<u8> {
-        (XDM, XDM_RELAY, XDM_LAST_CLEANUP_NONCE, chain_id, channel_id).encode()
+        (
+            prefix,
+            XDM,
+            XDM_RELAY,
+            XDM_LAST_CLEANUP_NONCE,
+            chain_id,
+            channel_id,
+        )
+            .encode()
     }
 
     pub(super) fn get_key_for_last_cleanup_relay_response_nonce(
+        prefix: &[u8],
         chain_id: ChainId,
         channel_id: ChannelId,
     ) -> Vec<u8> {
         (
+            prefix,
             XDM,
             XDM_RELAY_RESPONSE,
             XDM_LAST_CLEANUP_NONCE,
@@ -129,19 +155,19 @@ mod xdm_keys {
             .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(prefix: &[u8], id: MessageKey) -> Vec<u8> {
+        (prefix, XDM, XDM_RELAY, id).encode()
     }
 
-    pub(super) fn get_key_for_xdm_relay_response(id: MessageKey) -> Vec<u8> {
-        (XDM, XDM_RELAY_RESPONSE, id).encode()
+    pub(super) fn get_key_for_xdm_relay_response(prefix: &[u8], id: MessageKey) -> Vec<u8> {
+        (prefix, 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,
+pub struct BlockId<Block: BlockT> {
+    pub number: NumberFor<Block>,
+    pub hash: Block::Hash,
 }
 
 impl<Block: BlockT> From<Info<Block>> for BlockId<Block> {
@@ -156,32 +182,38 @@ impl<Block: BlockT> From<Info<Block>> for BlockId<Block> {
 /// Store the given XDM ID as processed at given block.
 pub fn set_xdm_message_processed_at<Backend, Block>(
     backend: &Backend,
+    prefix: &[u8],
     xdm_id: XdmId,
     block_id: BlockId<Block>,
 ) -> ClientResult<()>
 where
     Backend: AuxStore,
     Block: BlockT,
 {
-    let key = xdm_keys::get_key_for_xdm_id(xdm_id);
+    let key = xdm_keys::get_key_for_xdm_id(prefix, 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,
+    prefix: &[u8],
     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())
+    load_decode(
+        backend,
+        xdm_keys::get_key_for_xdm_id(prefix, xdm_id).as_slice(),
+    )
 }
 
 /// Cleans up all the xdm storages until the latest nonces.
 pub fn cleanup_chain_channel_storages<Backend>(
     backend: &Backend,
+    prefix: &[u8],
     chain_id: ChainId,
     channel_id: ChannelId,
     channel_nonce: ChannelNonce,
@@ -193,7 +225,7 @@ where
     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);
+            xdm_keys::get_key_for_last_cleanup_relay_nonce(prefix, chain_id, channel_id);
         let last_cleaned_up_nonce =
             load_decode::<_, Nonce>(backend, last_cleanup_relay_nonce_key.as_slice())?;
 
@@ -204,17 +236,19 @@ where
 
         tracing::debug!(
             target: LOG_TARGET,
-            "Cleaning Relay xdm keys for {:?} channel: {:?} from: {:?} to: {:?}",
+            "[{:?}]Cleaning Relay xdm keys for {:?} channel: {:?} from: {:?} to: {:?}",
+            to_hex(prefix, false),
             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,
-            )));
+            to_delete.push(xdm_keys::get_key_for_xdm_relay(
+                prefix,
+                (chain_id, channel_id, from_nonce),
+            ));
             from_nonce = from_nonce.saturating_add(Nonce::one());
         }
 
@@ -223,7 +257,7 @@ where
 
     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);
+            xdm_keys::get_key_for_last_cleanup_relay_response_nonce(prefix, chain_id, channel_id);
         let last_cleaned_up_nonce =
             load_decode::<_, Nonce>(backend, last_cleanup_relay_response_nonce_key.as_slice())?;
 
@@ -234,17 +268,19 @@ where
 
         tracing::debug!(
             target: LOG_TARGET,
-            "Cleaning Relay response xdm keys for {:?} channel: {:?} from: {:?} to: {:?}",
+            "[{:?}]Cleaning Relay response xdm keys for {:?} channel: {:?} from: {:?} to: {:?}",
+            to_hex(prefix, false),
             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,
-            )));
+            to_delete.push(xdm_keys::get_key_for_xdm_relay_response(
+                prefix,
+                (chain_id, channel_id, from_nonce),
+            ));
             from_nonce = from_nonce.saturating_add(Nonce::one());
         }
 
diff --git a/domains/client/cross-domain-message-gossip/src/lib.rs b/domains/client/cross-domain-message-gossip/src/lib.rs
@@ -5,9 +5,14 @@ mod aux_schema;
 mod gossip_worker;
 mod message_listener;
 
-pub use aux_schema::{get_channel_state, set_channel_state, ChannelDetail};
+pub use aux_schema::{
+    get_channel_state, get_xdm_processed_block_number, set_channel_state,
+    set_xdm_message_processed_at, BlockId, ChannelDetail,
+};
 pub use gossip_worker::{
     xdm_gossip_peers_set_config, ChainMsg, ChainSink, ChannelUpdate, GossipWorker,
     GossipWorkerBuilder, Message, MessageData,
 };
-pub use message_listener::start_cross_chain_message_listener;
+pub use message_listener::{
+    can_allow_xdm_submission, start_cross_chain_message_listener, RELAYER_PREFIX,
+};
diff --git a/domains/client/cross-domain-message-gossip/src/message_listener.rs b/domains/client/cross-domain-message-gossip/src/message_listener.rs
@@ -31,6 +31,9 @@ use subspace_runtime_primitives::{Balance, BlockNumber};
 use thiserror::Error;
 
 pub(crate) const LOG_TARGET: &str = "domain_message_listener";
+const TX_POOL_PREFIX: &[u8] = b"xdm_tx_pool_listener";
+pub const RELAYER_PREFIX: &[u8] = b"xdm_relayer";
+
 /// Number of blocks an already submitted XDM is not accepted since last submission.
 const XDM_ACCEPT_BLOCK_LIMIT: u32 = 5;
 
@@ -476,7 +479,7 @@ where
     Ok(())
 }
 
-fn can_allow_xdm_submission<Client, Block>(
+pub fn can_allow_xdm_submission<Client, Block>(
     client: &Arc<Client>,
     xdm_id: XdmId,
     submitted_block_id: BlockId<Block>,
@@ -564,7 +567,7 @@ where
             runtime_api.channel_nonce(block_id.hash, src_chain_id, channel_id)?;
 
         if let Some(submitted_block_id) =
-            get_xdm_processed_block_number::<_, BlockOf<TxPool>>(&**client, xdm_id)?
+            get_xdm_processed_block_number::<_, BlockOf<TxPool>>(&**client, TX_POOL_PREFIX, xdm_id)?
             && !can_allow_xdm_submission(
                 client,
                 xdm_id,
@@ -614,11 +617,17 @@ where
                 block_id
             );
 
-            set_xdm_message_processed_at(&**client, xdm_id, block_id)?;
+            set_xdm_message_processed_at(&**client, TX_POOL_PREFIX, xdm_id, block_id)?;
         }
 
         if let Some(channel_nonce) = maybe_channel_nonce {
-            cleanup_chain_channel_storages(&**client, src_chain_id, channel_id, channel_nonce)?;
+            cleanup_chain_channel_storages(
+                &**client,
+                TX_POOL_PREFIX,
+                src_chain_id,
+                channel_id,
+                channel_nonce,
+            )?;
         }
 
         Ok(true)
diff --git a/domains/client/relayer/src/lib.rs b/domains/client/relayer/src/lib.rs
