fix: hrmp

crates/pop-cli/src/commands/call/chain.rs

@@ -403,7 +403,7 @@ impl Call {
 			},
 		};
 		// If sudo is required, wrap the call in a sudo call.
-		let xt = if self.sudo { construct_sudo_extrinsic(xt)? } else { xt };
+		let xt = if self.sudo { construct_sudo_extrinsic(xt) } else { xt };
 		let encoded_data = encode_call_data(client, &xt)?;
 		// If the encoded call data is too long, don't display it all.
 		if encoded_data.len() < ENCODED_CALL_DATA_MAX_LEN {

crates/pop-cli/src/commands/up/parachain.rs

@@ -3,13 +3,13 @@
 use crate::style::{style, Theme};
 use clap::Args;
 use cliclack::{
-	clear_screen, confirm, intro, log, multi_progress, outro, outro_cancel, set_theme, ProgressBar,
-	Theme as _, ThemeState,
+	clear_screen, confirm, intro, log, multi_progress, outro, outro_cancel, set_theme, spinner,
+	ProgressBar, Theme as _, ThemeState,
 };
 use console::{Emoji, Style, Term};
 use duct::cmd;
 use pop_common::Status;
-use pop_parachains::{Error, IndexSet, NetworkNode, Zombienet};
+use pop_parachains::{clear_dmpq, Error, IndexSet, NetworkNode, RelayChain, Zombienet};
 use std::{path::Path, time::Duration};
 use tokio::time::sleep;
 
@@ -91,8 +91,8 @@
 		}
 
 		// Finally spawn network and wait for signal to terminate
-		let spinner = cliclack::spinner();
-		spinner.start("🚀 Launching local network...");
+		let progress = spinner();
+		progress.start("🚀 Launching local network...");
 		match zombienet.spawn().await {
 			Ok(network) => {
 				let mut result =
@@ -143,10 +143,39 @@
 				}
 
 				if let Some(command) = &self.command {
-					run_custom_command(&spinner, command).await?;
+					run_custom_command(&progress, command).await?;
+				}
+
+				progress.stop(result);
+
+				// Check for any specified channels
+				if zombienet.hrmp_channels() {
+					let relay_chain = zombienet.relay_chain();
+					match RelayChain::from(relay_chain) {
+						None => {
+							log::error(format!("🚫 Using `{relay_chain}` with HRMP channels is currently unsupported. Please use `paseo-local` or `westend-local`."))?;
+						},
+						Some(_) => {
+							let progress = spinner();
+							progress.start("Connecting to relay chain to prepare channels...");
+							// Allow relay node time to start
+							sleep(Duration::from_secs(10)).await;
+							progress.set_message("Preparing channels...");
+							let relay_endpoint = network.relaychain().nodes()[0].client().await?;
+							let para_ids: Vec<_> =
+								network.parachains().iter().map(|p| p.para_id()).collect();
+							tokio::spawn(async move {
+								if let Err(e) = clear_dmpq(relay_endpoint, &para_ids).await {
+									progress.stop(format!("🚫 Could not prepare channels: {e}"));
+									return Ok::<(), Error>(());
+								}
+								progress.stop("Channels successfully prepared for initialization.");
+								Ok::<(), Error>(())
+							});
+						},
+					}
 				}
 
-				spinner.stop(result);
 				tokio::signal::ctrl_c().await?;
 				outro("Done")?;
 			},
@@ -255,7 +284,7 @@
 		let binaries: Vec<_> = binaries
 			.into_iter()
 			.filter(|b| !b.exists() || (latest && b.stale()))
 			.map(|b| {
 				if latest && b.stale() {
 					b.use_latest()
 				}

crates/pop-parachains/Cargo.toml

@@ -17,6 +17,8 @@ glob.workspace = true
 serde_json.workspace = true
 strum.workspace = true
 strum_macros.workspace = true
+subxt-signer.workspace = true
+subxt.workspace = true
 tar.workspace = true
 tempfile.workspace = true
 thiserror.workspace = true
@@ -29,7 +31,6 @@ reqwest.workspace = true
 scale-info.workspace = true
 scale-value.workspace = true
 sp-core.workspace = true
-subxt.workspace = true
 symlink.workspace = true
 toml_edit.workspace = true
 walkdir.workspace = true

crates/pop-parachains/src/call/mod.rs

@@ -41,8 +41,8 @@ pub fn construct_extrinsic(
 /// # Arguments
 /// * `xt`: The extrinsic representing the dispatchable function call to be dispatched with `Root`
 ///   privileges.
-pub fn construct_sudo_extrinsic(xt: DynamicPayload) -> Result<DynamicPayload, Error> {
-	Ok(subxt::dynamic::tx("Sudo", "sudo", [xt.into_value()].to_vec()))
+pub fn construct_sudo_extrinsic(xt: DynamicPayload) -> DynamicPayload {
+	subxt::dynamic::tx("Sudo", "sudo", [xt.into_value()].to_vec())
 }
 
 /// Signs and submits a given extrinsic.
@@ -255,7 +255,7 @@ mod tests {
 				"100".to_string(),
 			],
 		)?;
-		let xt = construct_sudo_extrinsic(xt)?;
+		let xt = construct_sudo_extrinsic(xt);
 		assert_eq!(xt.call_name(), "sudo");
 		assert_eq!(xt.pallet_name(), "Sudo");
 		Ok(())

crates/pop-parachains/src/errors.rs

@@ -7,9 +7,9 @@
 #[derive(Error, Debug)]
 pub enum Error {
 	#[error("User aborted due to existing target directory.")]
 	Aborted,
 	#[error("Anyhow error: {0}")]
 	AnyhowError(#[from] anyhow::Error),
 	/// An error occurred while decoding the call data.
 	#[error("Failed to decode call data. {0}")]
 	CallDataDecodingError(String),
@@ -17,16 +17,16 @@
 	#[error("Failed to encode call data. {0}")]
 	CallDataEncodingError(String),
 	#[error("{0}")]
 	CommonError(#[from] pop_common::Error),
 	/// An error occurred while attempting to establish a connection to the endpoint.
 	#[error("Failed to establish a connection to: {0}")]
 	ConnectionFailure(String),
 	#[error("Configuration error: {0}")]
 	Config(String),
 	#[error("Failed to access the current directory")]
 	CurrentDirAccess,
 	#[error("Failed to parse the endowment value")]
 	EndowmentError,
 	/// An error occurred during the submission of an extrinsic.
 	#[error("Extrinsic submission error: {0}")]
 	ExtrinsicSubmissionError(String),
@@ -34,22 +34,22 @@
 	#[error("The dispatchable function is not supported")]
 	FunctionNotSupported,
 	#[error("IO error: {0}")]
 	IO(#[from] std::io::Error),
 	#[error("JSON error: {0}")]
 	JsonError(#[from] serde_json::Error),
 	/// An error occurred while parsing metadata of a parameter.
 	#[error("Error parsing metadata for parameter {0}")]
 	MetadataParsingError(String),
 	#[error("Missing binary: {0}")]
 	MissingBinary(String),
 	#[error("Missing chain spec file at: {0}")]
 	MissingChainSpec(String),
 	#[error("Command {command} doesn't exist in binary {binary}")]
 	MissingCommand { command: String, binary: String },
 	#[error("Orchestrator error: {0}")]
 	OrchestratorError(#[from] OrchestratorError),
 	#[error("Failed to create pallet directory")]
 	PalletDirCreation,
 	/// The specified pallet could not be found.
 	#[error("Failed to find the pallet {0}")]
 	PalletNotFound(String),
@@ -57,15 +57,18 @@
 	#[error("Failed to process the arguments provided by the user.")]
 	ParamProcessingError,
 	#[error("Invalid path")]
 	PathError,
 	#[error("Failed to execute rustfmt")]
 	RustfmtError(std::io::Error),
 	#[error("Template error: {0}")]
 	SourcingError(#[from] pop_common::sourcing::Error),
+	/// An error occurred whilst interacting with a chain using `subxt`.
+	#[error("Subxt error: {0}")]
+	SubXtError(#[from] subxt::Error),
 	#[error("Toml error: {0}")]
 	TomlError(#[from] toml_edit::de::Error),
 	#[error("Unsupported command: {0}")]
 	UnsupportedCommand(String),
 	#[error("Failed to locate the workspace")]
 	WorkspaceLocate,
 }

crates/pop-parachains/src/lib.rs

@@ -8,6 +8,7 @@ mod errors;
 mod generator;
 mod new_pallet;
 mod new_parachain;
+mod relay;
 mod templates;
 mod up;
 mod utils;
@@ -30,6 +31,7 @@ pub use errors::Error;
 pub use indexmap::IndexSet;
 pub use new_pallet::{create_pallet_template, new_pallet_options::*, TemplatePalletConfig};
 pub use new_parachain::instantiate_template_dir;
+pub use relay::{clear_dmpq, RelayChain};
 // External export from subxt.
 pub use subxt::{
 	tx::{DynamicPayload, Payload},

crates/pop-parachains/src/relay.rs

@@ -0,0 +1,146 @@
+// SPDX-License-Identifier: GPL-3.0
+
+use crate::{call, DynamicPayload, Error};
+use sp_core::twox_128;
+use subxt::{
+	config::BlockHash,
+	dynamic::{self, Value},
+	ext::sp_core,
+	OnlineClient, PolkadotConfig,
+};
+
+/// Clears the DMPQ state for the given parachain IDs.
+///
+/// # Arguments
+/// * `client` - Client for the network which state is to be modified.
+/// * `para_ids` - List of ids to build the keys that will be mutated.
+pub async fn clear_dmpq(
+	client: OnlineClient<PolkadotConfig>,
+	para_ids: &[u32],
+) -> Result<impl BlockHash, Error> {
+	// Wait for blocks to be produced.
+	let mut sub = client.blocks().subscribe_finalized().await?;
+	for _ in 0..2 {
+		sub.next().await;
+	}
+
+	// Generate storage keys to be removed
+	let clear_dmq_keys = generate_storage_keys(para_ids);
+
+	// Submit calls to remove specified keys
+	let kill_storage = construct_kill_storage_call(clear_dmq_keys);
+	let sudo = subxt_signer::sr25519::dev::alice();
+	let sudo_call = call::construct_sudo_extrinsic(kill_storage);
+	Ok(client.tx().sign_and_submit_default(&sudo_call, &sudo).await?)
+}
+
+fn construct_kill_storage_call(keys: Vec<Vec<u8>>) -> DynamicPayload {
+	dynamic::tx(
+		"System",
+		"kill_storage",
+		vec![Value::unnamed_composite(keys.into_iter().map(Value::from_bytes))],
+	)
+}
+
+fn generate_storage_keys(para_ids: &[u32]) -> Vec<Vec<u8>> {
+	let dmp = twox_128("Dmp".as_bytes());
+	let dmp_queue_heads = twox_128("DownwardMessageQueueHeads".as_bytes());
+	let dmp_queues = twox_128("DownwardMessageQueues".as_bytes());
+	let mut clear_dmq_keys = Vec::<Vec<u8>>::new();
+	for id in para_ids {
+		let id = id.to_le_bytes();
+		// DMP Queue Head
+		let mut key = dmp.to_vec();
+		key.extend(&dmp_queue_heads);
+		key.extend(sp_core::twox_64(&id));
+		key.extend(id);
+		clear_dmq_keys.push(key);
+		// DMP Queue
+		let mut key = dmp.to_vec();
+		key.extend(&dmp_queues);
+		key.extend(sp_core::twox_64(&id));
+		key.extend(id);
+		clear_dmq_keys.push(key);
+	}
+	clear_dmq_keys
+}
+
+/// A supported relay chain.
+#[derive(Debug, PartialEq)]
+pub enum RelayChain {
+	/// Paseo.
+	PaseoLocal,
+	/// Westend.
+	WestendLocal,
+}
+
+impl RelayChain {
+	/// Attempts to convert a chain identifier into a supported `RelayChain` variant.
+	///
+	/// # Arguments
+	/// * `id` - The relay chain identifier.
+	pub fn from(id: &str) -> Option<RelayChain> {
+		match id {
+			"paseo-local" => Some(RelayChain::PaseoLocal),
+			"westend-local" => Some(RelayChain::WestendLocal),
+			_ => None,
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use subxt::ext::sp_core::twox_64;
+	use RelayChain::*;
+
+	#[test]
+	fn construct_kill_storage_call_works() {
+		let keys = vec!["key".as_bytes().to_vec()];
+		assert_eq!(
+			construct_kill_storage_call(keys.clone()),
+			dynamic::tx(
+				"System",
+				"kill_storage",
+				vec![Value::unnamed_composite(keys.into_iter().map(Value::from_bytes))],
+			)
+		)
+	}
+
+	#[test]
+	fn generate_storage_keys_works() {
+		let para_ids = vec![1_000, 4_385];
+		let dmp = twox_128("Dmp".as_bytes());
+		let dmp_queue_heads = [dmp, twox_128("DownwardMessageQueueHeads".as_bytes())].concat();
+		let dmp_queues = [dmp, twox_128("DownwardMessageQueues".as_bytes())].concat();
+
+		assert_eq!(
+			generate_storage_keys(&para_ids),
+			para_ids
+				.iter()
+				.flat_map(|id| {
+					let id = id.to_le_bytes().to_vec();
+					[
+						// DMP Queue Head
+						[dmp_queue_heads.clone(), twox_64(&id).to_vec(), id.clone()].concat(),
+						// DMP Queue
+						[dmp_queues.clone(), twox_64(&id).to_vec(), id].concat(),
+					]
+				})
+				.collect::<Vec<_>>()
+		)
+	}
+
+	#[test]
+	fn supported_relay_chains() {
+		for (s, e) in [
+			// Only chains with sudo supported
+			("paseo-local", Some(PaseoLocal)),
+			("westend-local", Some(WestendLocal)),
+			("kusama-local", None),
+			("polkadot-local", None),
+		] {
+			assert_eq!(RelayChain::from(s), e)
+		}
+	}
+}