Fix auto-pause xcm: incoming XCMP messages where dropped when auto-pause

Cargo.toml

@@ -2,8 +2,8 @@
 exclude = ["bin/utils/moonkey"]
 members = [
 	"bin/utils/moonkey",
+	"client/rpc/dev",
 	"client/rpc/finality",
-	"client/rpc/manual-xcm",
 	"client/vrf",
 	"node",
 	"node/cli",
@@ -117,8 +117,8 @@ moonbeam-cli = { path = "node/cli", default-features = false }
 moonbeam-cli-opt = { path = "node/cli-opt", default-features = false }
 moonbeam-service = { path = "node/service", default-features = false }
 
-manual-xcm-rpc = { path = "client/rpc/manual-xcm" }
 moonbeam-client-evm-tracing = { path = "client/evm-tracing" }
+moonbeam-dev-rpc = { path = "client/rpc/dev" }
 moonbeam-finality-rpc = { path = "client/rpc/finality" }
 moonbeam-rpc-core-debug = { path = "client/rpc-core/debug" }
 moonbeam-rpc-core-trace = { path = "client/rpc-core/trace" }

client/rpc/manual-xcm/Cargo.toml → client/rpc/dev/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "manual-xcm-rpc"
+name = "moonbeam-dev-rpc"
 authors = { workspace = true }
 edition = "2021"
 homepage = "https://moonbeam.network"

client/rpc/manual-xcm/src/lib.rs → client/rpc/dev/src/lib.rs

@@ -13,6 +13,7 @@
 
 // You should have received a copy of the GNU General Public License
 // along with Moonbeam.  If not, see <http://www.gnu.org/licenses/>.
+
 use cumulus_primitives_core::ParaId;
 use cumulus_primitives_core::XcmpMessageFormat;
 use jsonrpsee::{
@@ -28,12 +29,10 @@ use xcm::opaque::lts::Weight;
 use xcm::v4::prelude::*;
 use xcm_primitives::DEFAULT_PROOF_SIZE;
 
-/// This RPC interface is used to manually submit XCM messages that will be injected into a
-/// parachain-enabled runtime. This allows testing XCM logic in a controlled way in integration
-/// tests.
+/// This RPC interface is used to provide methods in dev mode only
 #[rpc(server)]
 #[jsonrpsee::core::async_trait]
-pub trait ManualXcmApi {
+pub trait DevApi {
 	/// Inject a downward xcm message - A message that comes from the relay chain.
 	/// You may provide an arbitrary message, or if you provide an emtpy byte array,
 	/// Then a default message (DOT transfer down to ALITH) will be injected
@@ -53,15 +52,20 @@ pub trait ManualXcmApi {
 	/// transfer of the sending paraId's native token will be injected.
 	#[method(name = "xcm_injectHrmpMessage")]
 	async fn inject_hrmp_message(&self, sender: ParaId, message: Vec<u8>) -> RpcResult<()>;
+
+	/// Skip N relay blocks, for testing purposes
+	#[method(name = "test_skipRelayBlocks")]
+	async fn skip_relay_blocks(&self, n: u32) -> RpcResult<()>;
 }
 
-pub struct ManualXcm {
+pub struct DevRpc {
 	pub downward_message_channel: flume::Sender<Vec<u8>>,
 	pub hrmp_message_channel: flume::Sender<(ParaId, Vec<u8>)>,
+	pub additional_relay_offset: std::sync::Arc<std::sync::atomic::AtomicU32>,
 }
 
 #[jsonrpsee::core::async_trait]
-impl ManualXcmApiServer for ManualXcm {
+impl DevApiServer for DevRpc {
 	async fn inject_downward_message(&self, msg: Vec<u8>) -> RpcResult<()> {
 		let downward_message_channel = self.downward_message_channel.clone();
 		// If no message is supplied, inject a default one.
@@ -148,6 +152,12 @@ impl ManualXcmApiServer for ManualXcm {
 
 		Ok(())
 	}
+
+	async fn skip_relay_blocks(&self, n: u32) -> RpcResult<()> {
+		self.additional_relay_offset
+			.fetch_add(n, std::sync::atomic::Ordering::SeqCst);
+		Ok(())
+	}
 }
 
 // This bit cribbed from frontier.

node/service/Cargo.toml

@@ -28,7 +28,7 @@ tokio = { workspace = true, features = ["macros", "sync"] }
 trie-root = { workspace = true }
 
 # Moonbeam
-manual-xcm-rpc = { workspace = true }
+moonbeam-dev-rpc = { workspace = true }
 moonbeam-cli-opt = { workspace = true }
 moonbeam-core-primitives = { workspace = true }
 moonbeam-finality-rpc = { workspace = true }

node/service/src/lib.rs

@@ -774,7 +774,7 @@ where
 				fee_history_cache: fee_history_cache.clone(),
 				network: network.clone(),
 				sync: sync.clone(),
-				xcm_senders: None,
+				dev_rpc_data: None,
 				block_data_cache: block_data_cache.clone(),
 				overrides: overrides.clone(),
 				forced_parent_hashes,
@@ -1204,7 +1204,7 @@ where
 	let overrides = Arc::new(StorageOverrideHandler::new(client.clone()));
 	let fee_history_limit = rpc_config.fee_history_limit;
 	let mut command_sink = None;
-	let mut xcm_senders = None;
+	let mut dev_rpc_data = None;
 	let collator = config.role.is_authority();
 
 	if collator {
@@ -1269,7 +1269,12 @@ where
 		// Create channels for mocked XCM messages.
 		let (downward_xcm_sender, downward_xcm_receiver) = flume::bounded::<Vec<u8>>(100);
 		let (hrmp_xcm_sender, hrmp_xcm_receiver) = flume::bounded::<(ParaId, Vec<u8>)>(100);
-		xcm_senders = Some((downward_xcm_sender, hrmp_xcm_sender));
+		let additional_relay_offset = Arc::new(std::sync::atomic::AtomicU32::new(0));
+		dev_rpc_data = Some((
+			downward_xcm_sender,
+			hrmp_xcm_sender,
+			additional_relay_offset.clone(),
+		));
 
 		let client_clone = client.clone();
 		let keystore_clone = keystore_container.keystore().clone();
@@ -1304,6 +1309,7 @@ where
 					let maybe_current_para_head = client_set_aside_for_cidp.expect_header(block);
 					let downward_xcm_receiver = downward_xcm_receiver.clone();
 					let hrmp_xcm_receiver = hrmp_xcm_receiver.clone();
+					let additional_relay_offset = additional_relay_offset.clone();
 
 					let client_for_xcm = client_set_aside_for_cidp.clone();
 					async move {
@@ -1324,7 +1330,8 @@ where
 						let mocked_parachain = MockValidationDataInherentDataProvider {
 							current_para_block,
 							current_para_block_head,
-							relay_offset: 1000,
+							relay_offset: 1000
+								+ additional_relay_offset.load(std::sync::atomic::Ordering::SeqCst),
 							relay_blocks_per_para_block: 2,
 							// TODO: Recheck
 							para_blocks_per_relay_epoch: 10,
@@ -1440,7 +1447,7 @@ where
 				fee_history_cache: fee_history_cache.clone(),
 				network: network.clone(),
 				sync: sync.clone(),
-				xcm_senders: xcm_senders.clone(),
+				dev_rpc_data: dev_rpc_data.clone(),
 				overrides: overrides.clone(),
 				block_data_cache: block_data_cache.clone(),
 				forced_parent_hashes: None,