Relay subcommand that performs token RLT <> MLAU token swap (#1141)

* token swap relay

* token swap subcommand fixes

* fmt

* removed debug traces

* removed commented code

bridges/bin/millau/node/src/chain_spec.rs

@@ -135,6 +135,7 @@ impl Alternative {
 							get_account_id_from_seed::<sr25519::Public>("George//stash"),
 							get_account_id_from_seed::<sr25519::Public>("Harry//stash"),
 							get_account_id_from_seed::<sr25519::Public>("RialtoMessagesOwner"),
+							get_account_id_from_seed::<sr25519::Public>("WithRialtoTokenSwap"),
 							pallet_bridge_messages::relayer_fund_account_id::<
 								bp_millau::AccountId,
 								bp_millau::AccountIdConverter,

bridges/bin/rialto/node/src/chain_spec.rs

@@ -151,6 +151,7 @@ impl Alternative {
 							get_account_id_from_seed::<sr25519::Public>("George//stash"),
 							get_account_id_from_seed::<sr25519::Public>("Harry//stash"),
 							get_account_id_from_seed::<sr25519::Public>("MillauMessagesOwner"),
+							get_account_id_from_seed::<sr25519::Public>("WithMillauTokenSwap"),
 							pallet_bridge_messages::relayer_fund_account_id::<
 								bp_rialto::AccountId,
 								bp_rialto::AccountIdConverter,

bridges/modules/messages/src/lib.rs

@@ -777,12 +777,11 @@ pub mod pallet {
 /// messages and lanes states proofs.
 pub mod storage_keys {
 	use super::*;
-	use frame_support::StorageHasher;
 	use sp_core::storage::StorageKey;
 
 	/// Storage key of the outbound message in the runtime storage.
 	pub fn message_key(pallet_prefix: &str, lane: &LaneId, nonce: MessageNonce) -> StorageKey {
-		storage_map_final_key(
+		bp_runtime::storage_map_final_key_blake2_128concat(
 			pallet_prefix,
 			"OutboundMessages",
 			&MessageKey { lane_id: *lane, nonce }.encode(),
@@ -791,28 +790,12 @@ pub mod storage_keys {
 
 	/// Storage key of the outbound message lane state in the runtime storage.
 	pub fn outbound_lane_data_key(pallet_prefix: &str, lane: &LaneId) -> StorageKey {
-		storage_map_final_key(pallet_prefix, "OutboundLanes", lane)
+		bp_runtime::storage_map_final_key_blake2_128concat(pallet_prefix, "OutboundLanes", lane)
 	}
 
 	/// Storage key of the inbound message lane state in the runtime storage.
 	pub fn inbound_lane_data_key(pallet_prefix: &str, lane: &LaneId) -> StorageKey {
-		storage_map_final_key(pallet_prefix, "InboundLanes", lane)
-	}
-
-	/// This is a copypaste of the `frame_support::storage::generator::StorageMap::storage_map_final_key`.
-	fn storage_map_final_key(pallet_prefix: &str, map_name: &str, key: &[u8]) -> StorageKey {
-		let pallet_prefix_hashed = frame_support::Twox128::hash(pallet_prefix.as_bytes());
-		let storage_prefix_hashed = frame_support::Twox128::hash(map_name.as_bytes());
-		let key_hashed = frame_support::Blake2_128Concat::hash(key);
-
-		let mut final_key =
-			Vec::with_capacity(pallet_prefix_hashed.len() + storage_prefix_hashed.len() + key_hashed.len());
-
-		final_key.extend_from_slice(&pallet_prefix_hashed[..]);
-		final_key.extend_from_slice(&storage_prefix_hashed[..]);
-		final_key.extend_from_slice(key_hashed.as_ref());
-
-		StorageKey(final_key)
+		bp_runtime::storage_map_final_key_blake2_128concat(pallet_prefix, "InboundLanes", lane)
 	}
 }
 

bridges/modules/token-swap/src/lib.rs

@@ -56,12 +56,11 @@ use bp_messages::{
 	DeliveredMessages, LaneId, MessageNonce,
 };
 use bp_runtime::{messages::DispatchFeePayment, ChainId};
-use bp_token_swap::{TokenSwap, TokenSwapType};
-use codec::{Decode, Encode};
+use bp_token_swap::{TokenSwap, TokenSwapState, TokenSwapType};
+use codec::Encode;
 use frame_support::{
 	fail,
 	traits::{Currency, ExistenceRequirement},
-	RuntimeDebug,
 };
 use sp_core::H256;
 use sp_io::hashing::blake2_256;
@@ -71,22 +70,11 @@ use sp_std::vec::Vec;
 #[cfg(test)]
 mod mock;
 
-/// Pending token swap state.
-#[derive(Encode, Decode, Clone, RuntimeDebug, PartialEq, Eq)]
-pub enum TokenSwapState {
-	/// The swap has been started using the `start_claim` call, but we have no proof that it has
-	/// happened at the Bridged chain.
-	Started,
-	/// The swap has happened at the Bridged chain and may be claimed by the Bridged chain party using
-	/// the `claim_swap` call.
-	Confirmed,
-	/// The swap has failed at the Bridged chain and This chain party may cancel it using the
-	/// `cancel_swap` call.
-	Failed,
-}
-
 pub use pallet::*;
 
+/// Name of the `PendingSwaps` storage map.
+pub const PENDING_SWAPS_MAP_NAME: &str = "PendingSwaps";
+
 // comes from #[pallet::event]
 #[allow(clippy::unused_unit)]
 #[frame_support::pallet]
@@ -324,6 +312,13 @@ pub mod pallet {
 					return sp_runtime::TransactionOutcome::Rollback(Err(Error::<T, I>::SwapAlreadyStarted.into()));
 				}
 
+				log::trace!(
+					target: "runtime::bridge-token-swap",
+					"The swap {:?} (hash {:?}) has been started",
+					swap,
+					swap_hash,
+				);
+
 				// remember that we're waiting for the transfer message delivery confirmation
 				PendingMessages::<T, I>::insert(transfer_message_nonce, swap_hash);
 
@@ -475,14 +470,21 @@ pub mod pallet {
 				reads += 1;
 				if let Some(swap_hash) = PendingMessages::<T, I>::take(message_nonce) {
 					writes += 1;
-					PendingSwaps::<T, I>::insert(
+
+					let token_swap_state = if delivered_messages.message_dispatch_result(message_nonce) {
+						TokenSwapState::Confirmed
+					} else {
+						TokenSwapState::Failed
+					};
+
+					log::trace!(
+						target: "runtime::bridge-token-swap",
+						"The dispatch of swap {:?} has been completed with {:?} status",
 						swap_hash,
-						if delivered_messages.message_dispatch_result(message_nonce) {
-							TokenSwapState::Confirmed
-						} else {
-							TokenSwapState::Failed
-						},
+						token_swap_state,
 					);
+
+					PendingSwaps::<T, I>::insert(swap_hash, token_swap_state);
 				}
 			}
 
@@ -534,6 +536,18 @@ pub mod pallet {
 				));
 			}
 
+			log::trace!(
+				target: "runtime::bridge-token-swap",
+				"The swap {:?} (hash {:?}) has been completed with {} status",
+				swap,
+				swap_hash,
+				match event {
+					Event::SwapClaimed(_) => "claimed",
+					Event::SwapCancelled(_) => "cancelled",
+					_ => "<unknown>",
+				},
+			);
+
 			// forget about swap
 			PendingSwaps::<T, I>::remove(swap_hash);
 

bridges/primitives/chain-millau/src/lib.rs

@@ -258,6 +258,8 @@ pub fn max_extrinsic_size() -> u32 {
 
 /// Name of the With-Rialto messages pallet instance in the Millau runtime.
 pub const WITH_RIALTO_MESSAGES_PALLET_NAME: &str = "BridgeRialtoMessages";
+/// Name of the With-Rialto token swap pallet instance in the Millau runtime.
+pub const WITH_RIALTO_TOKEN_SWAP_PALLET_NAME: &str = "BridgeRialtoTokenSwap";
 
 /// Name of the `MillauFinalityApi::best_finalized` runtime method.
 pub const BEST_FINALIZED_MILLAU_HEADER_METHOD: &str = "MillauFinalityApi_best_finalized";

bridges/primitives/runtime/src/lib.rs

@@ -19,7 +19,7 @@
 #![cfg_attr(not(feature = "std"), no_std)]
 
 use codec::Encode;
-use frame_support::RuntimeDebug;
+use frame_support::{RuntimeDebug, StorageHasher};
 use sp_core::{hash::H256, storage::StorageKey};
 use sp_io::hashing::blake2_256;
 use sp_std::{convert::TryFrom, vec::Vec};
@@ -184,6 +184,33 @@ impl<BlockNumber: Copy + Into<u64>, BlockHash: Copy> TransactionEra<BlockNumber,
 	}
 }
 
+/// This is a copypaste of the `frame_support::storage::generator::StorageMap::storage_map_final_key`
+/// for `Blake2_128Concat` maps.
+///
+/// We're using it because to call `storage_map_final_key` directly, we need access to the runtime
+/// and pallet instance, which (sometimes) is impossible.
+pub fn storage_map_final_key_blake2_128concat(pallet_prefix: &str, map_name: &str, key: &[u8]) -> StorageKey {
+	storage_map_final_key_identity(pallet_prefix, map_name, &frame_support::Blake2_128Concat::hash(key))
+}
+
+/// This is a copypaste of the `frame_support::storage::generator::StorageMap::storage_map_final_key`
+/// for `Identity` maps.
+///
+/// We're using it because to call `storage_map_final_key` directly, we need access to the runtime
+/// and pallet instance, which (sometimes) is impossible.
+pub fn storage_map_final_key_identity(pallet_prefix: &str, map_name: &str, key_hashed: &[u8]) -> StorageKey {
+	let pallet_prefix_hashed = frame_support::Twox128::hash(pallet_prefix.as_bytes());
+	let storage_prefix_hashed = frame_support::Twox128::hash(map_name.as_bytes());
+
+	let mut final_key = Vec::with_capacity(pallet_prefix_hashed.len() + storage_prefix_hashed.len() + key_hashed.len());
+
+	final_key.extend_from_slice(&pallet_prefix_hashed[..]);
+	final_key.extend_from_slice(&storage_prefix_hashed[..]);
+	final_key.extend_from_slice(key_hashed.as_ref());
+
+	StorageKey(final_key)
+}
+
 /// This is how a storage key of storage parameter (`parameter_types! { storage Param: bool = false; }`) is computed.
 ///
 /// Copypaste from `frame_support::parameter_types` macro

bridges/primitives/token-swap/src/lib.rs

@@ -20,6 +20,20 @@ use codec::{Decode, Encode};
 use frame_support::RuntimeDebug;
 use sp_core::U256;
 
+/// Pending token swap state.
+#[derive(Encode, Decode, Clone, RuntimeDebug, PartialEq, Eq)]
+pub enum TokenSwapState {
+	/// The swap has been started using the `start_claim` call, but we have no proof that it has
+	/// happened at the Bridged chain.
+	Started,
+	/// The swap has happened at the Bridged chain and may be claimed by the Bridged chain party using
+	/// the `claim_swap` call.
+	Confirmed,
+	/// The swap has failed at the Bridged chain and This chain party may cancel it using the
+	/// `cancel_swap` call.
+	Failed,
+}
+
 /// Token swap type.
 ///
 /// Different swap types give a different guarantees regarding possible swap

bridges/relays/bin-substrate/Cargo.toml

@@ -15,6 +15,7 @@ log = "0.4.14"
 num-format = "0.4"
 num-traits = "0.2"
 paste = "1.0"
+rand = "0.8"
 structopt = "0.3"
 strum = { version = "0.21.0", features = ["derive"] }
 
@@ -28,6 +29,7 @@ bp-millau = { path = "../../primitives/chain-millau" }
 bp-polkadot = { path = "../../primitives/chain-polkadot" }
 bp-rialto = { path = "../../primitives/chain-rialto" }
 bp-rococo = { path = "../../primitives/chain-rococo" }
+bp-token-swap = { path = "../../primitives/token-swap" }
 bp-wococo = { path = "../../primitives/chain-wococo" }
 bp-runtime = { path = "../../primitives/runtime" }
 bp-westend = { path = "../../primitives/chain-westend" }
@@ -35,7 +37,9 @@ bridge-runtime-common = { path = "../../bin/runtime-common" }
 finality-relay = { path = "../finality" }
 messages-relay = { path = "../messages" }
 millau-runtime = { path = "../../bin/millau/runtime" }
+pallet-bridge-dispatch = { path = "../../modules/dispatch" }
 pallet-bridge-messages = { path = "../../modules/messages" }
+pallet-bridge-token-swap = { path = "../../modules/token-swap" }
 relay-kusama-client = { path = "../client-kusama" }
 relay-millau-client = { path = "../client-millau" }
 relay-polkadot-client = { path = "../client-polkadot" }
@@ -51,6 +55,7 @@ substrate-relay-helper = { path = "../lib-substrate-relay" }
 # Substrate Dependencies
 
 frame-support = { git = "https://github.com/paritytech/substrate", branch = "master" }
+pallet-balances = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-io = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }

bridges/relays/bin-substrate/src/cli/mod.rs

@@ -36,6 +36,7 @@ mod relay_headers;
 mod relay_headers_and_messages;
 mod relay_messages;
 mod resubmit_transactions;
+mod swap_tokens;
 
 /// Parse relay CLI args.
 pub fn parse_args() -> Command {
@@ -89,6 +90,8 @@ pub enum Command {
 	DeriveAccount(derive_account::DeriveAccount),
 	/// Resubmit transactions with increased tip if they are stalled.
 	ResubmitTransactions(resubmit_transactions::ResubmitTransactions),
+	/// Swap tokens using token-swap bridge.
+	SwapTokens(swap_tokens::SwapTokens),
 }
 
 impl Command {
@@ -120,6 +123,7 @@ impl Command {
 			Self::EstimateFee(arg) => arg.run().await?,
 			Self::DeriveAccount(arg) => arg.run().await?,
 			Self::ResubmitTransactions(arg) => arg.run().await?,
+			Self::SwapTokens(arg) => arg.run().await?,
 		}
 		Ok(())
 	}

bridges/relays/bin-substrate/src/cli/swap_tokens.rs

@@ -0,0 +1,640 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Tokens swap using token-swap bridge pallet.
+
+use codec::Encode;
+use num_traits::One;
+use rand::random;
+use structopt::StructOpt;
+use strum::{EnumString, EnumVariantNames, VariantNames};
+
+use frame_support::dispatch::GetDispatchInfo;
+use relay_substrate_client::{
+	AccountIdOf, AccountPublicOf, BalanceOf, BlockNumberOf, CallOf, Chain, ChainWithBalances, Client,
+	Error as SubstrateError, HashOf, SignatureOf, Subscription, TransactionSignScheme, TransactionStatusOf,
+	UnsignedTransaction,
+};
+use sp_core::{blake2_256, storage::StorageKey, Bytes, Pair, H256, U256};
+use sp_runtime::traits::{Convert, Header as HeaderT};
+
+use crate::cli::{
+	Balance, CliChain, SourceConnectionParams, SourceSigningParams, TargetConnectionParams, TargetSigningParams,
+};
+
+/// Swap tokens.
+#[derive(StructOpt)]
+pub struct SwapTokens {
+	/// A bridge instance to use in token swap.
+	#[structopt(possible_values = SwapTokensBridge::VARIANTS, case_insensitive = true)]
+	bridge: SwapTokensBridge,
+
+	#[structopt(flatten)]
+	source: SourceConnectionParams,
+	#[structopt(flatten)]
+	source_sign: SourceSigningParams,
+
+	#[structopt(flatten)]
+	target: TargetConnectionParams,
+	#[structopt(flatten)]
+	target_sign: TargetSigningParams,
+
+	#[structopt(subcommand)]
+	swap_type: TokenSwapType,
+	/// Source chain balance that source signer wants to swap.
+	#[structopt(long)]
+	source_balance: Balance,
+	/// Target chain balance that target signer wants to swap.
+	#[structopt(long)]
+	target_balance: Balance,
+}
+
+/// Token swap type.
+#[derive(StructOpt, Debug, PartialEq, Eq, Clone)]
+pub enum TokenSwapType {
+	/// The `target_sign` is temporary and only have funds for single swap.
+	NoLock,
+	/// This swap type prevents `source_signer` from restarting the swap after it has been completed.
+	LockUntilBlock {
+		/// Number of blocks before the swap expires.
+		#[structopt(long)]
+		blocks_before_expire: u32,
+		/// Unique swap nonce.
+		#[structopt(long)]
+		swap_nonce: Option<U256>,
+	},
+}
+
+/// Swap tokens bridge.
+#[derive(Debug, EnumString, EnumVariantNames)]
+#[strum(serialize_all = "kebab_case")]
+pub enum SwapTokensBridge {
+	/// Use token-swap pallet deployed at Millau to swap tokens with Rialto.
+	MillauToRialto,
+}
+
+macro_rules! select_bridge {
+	($bridge: expr, $generic: tt) => {
+		match $bridge {
+			SwapTokensBridge::MillauToRialto => {
+				type Source = relay_millau_client::Millau;
+				type Target = relay_rialto_client::Rialto;
+
+				type FromSwapToThisAccountIdConverter = bp_rialto::AccountIdConverter;
+
+				use bp_millau::{
+					derive_account_from_rialto_id as derive_source_account_from_target_account,
+					TO_MILLAU_ESTIMATE_MESSAGE_FEE_METHOD as ESTIMATE_TARGET_TO_SOURCE_MESSAGE_FEE_METHOD,
+					WITH_RIALTO_TOKEN_SWAP_PALLET_NAME as TOKEN_SWAP_PALLET_NAME,
+				};
+				use bp_rialto::{
+					derive_account_from_millau_id as derive_target_account_from_source_account,
+					TO_RIALTO_ESTIMATE_MESSAGE_FEE_METHOD as ESTIMATE_SOURCE_TO_TARGET_MESSAGE_FEE_METHOD,
+				};
+
+				const SOURCE_CHAIN_ID: bp_runtime::ChainId = bp_runtime::MILLAU_CHAIN_ID;
+				const TARGET_CHAIN_ID: bp_runtime::ChainId = bp_runtime::RIALTO_CHAIN_ID;
+
+				const SOURCE_SPEC_VERSION: u32 = millau_runtime::VERSION.spec_version;
+				const TARGET_SPEC_VERSION: u32 = rialto_runtime::VERSION.spec_version;
+
+				const SOURCE_TO_TARGET_LANE_ID: bp_messages::LaneId = *b"swap";
+				const TARGET_TO_SOURCE_LANE_ID: bp_messages::LaneId = [0, 0, 0, 0];
+
+				$generic
+			}
+		}
+	};
+}
+
+impl SwapTokens {
+	/// Run the command.
+	pub async fn run(self) -> anyhow::Result<()> {
+		select_bridge!(self.bridge, {
+			let source_client = self.source.to_client::<Source>().await?;
+			let source_sign = self.source_sign.to_keypair::<Target>()?;
+			let target_client = self.target.to_client::<Target>().await?;
+			let target_sign = self.target_sign.to_keypair::<Target>()?;
+
+			// names of variables in this function are matching names used by the `pallet-bridge-token-swap`
+
+			// prepare token swap intention
+			let token_swap = self
+				.prepare_token_swap::<Source, Target>(&source_client, &source_sign, &target_sign)
+				.await?;
+
+			// group all accounts that will be used later
+			let accounts = TokenSwapAccounts {
+				source_account_at_bridged_chain: derive_target_account_from_source_account(
+					bp_runtime::SourceAccount::Account(token_swap.source_account_at_this_chain.clone()),
+				),
+				target_account_at_this_chain: derive_source_account_from_target_account(
+					bp_runtime::SourceAccount::Account(token_swap.target_account_at_bridged_chain.clone()),
+				),
+				source_account_at_this_chain: token_swap.source_account_at_this_chain.clone(),
+				target_account_at_bridged_chain: token_swap.target_account_at_bridged_chain.clone(),
+				swap_account: FromSwapToThisAccountIdConverter::convert(token_swap.using_encoded(blake2_256).into()),
+			};
+
+			// account balances are used to demonstrate what's happening :)
+			let initial_balances = read_account_balances(&accounts, &source_client, &target_client).await?;
+
+			// before calling something that may fail, log what we're trying to do
+			log::info!(target: "bridge", "Starting swap: {:?}", token_swap);
+			log::info!(target: "bridge", "Swap accounts: {:?}", accounts);
+			log::info!(target: "bridge", "Initial account balances: {:?}", initial_balances);
+
+			//
+			// Step 1: swap is created
+			//
+
+			// prepare `Currency::transfer` call that will happen at the target chain
+			let bridged_currency_transfer: CallOf<Target> = pallet_balances::Call::transfer(
+				accounts.source_account_at_bridged_chain.clone().into(),
+				token_swap.target_balance_at_bridged_chain,
+			)
+			.into();
+			let bridged_currency_transfer_weight = bridged_currency_transfer.get_dispatch_info().weight;
+
+			// sign message
+			let bridged_chain_spec_version = TARGET_SPEC_VERSION;
+			let signature_payload = pallet_bridge_dispatch::account_ownership_digest(
+				&bridged_currency_transfer,
+				&accounts.swap_account,
+				&bridged_chain_spec_version,
+				SOURCE_CHAIN_ID,
+				TARGET_CHAIN_ID,
+			);
+			let bridged_currency_transfer_signature: SignatureOf<Target> = target_sign.sign(&signature_payload).into();
+
+			// prepare `create_swap` call
+			let target_public_at_bridged_chain: AccountPublicOf<Target> = target_sign.public().into();
+			let swap_delivery_and_dispatch_fee: BalanceOf<Source> =
+				crate::cli::estimate_fee::estimate_message_delivery_and_dispatch_fee(
+					&source_client,
+					ESTIMATE_SOURCE_TO_TARGET_MESSAGE_FEE_METHOD,
+					SOURCE_TO_TARGET_LANE_ID,
+					bp_message_dispatch::MessagePayload {
+						spec_version: TARGET_SPEC_VERSION,
+						weight: bridged_currency_transfer_weight,
+						origin: bp_message_dispatch::CallOrigin::TargetAccount(
+							accounts.swap_account.clone(),
+							target_public_at_bridged_chain.clone(),
+							bridged_currency_transfer_signature.clone(),
+						),
+						dispatch_fee_payment: bp_runtime::messages::DispatchFeePayment::AtTargetChain,
+						call: bridged_currency_transfer.encode(),
+					},
+				)
+				.await?;
+			let create_swap_call: CallOf<Source> = pallet_bridge_token_swap::Call::create_swap(
+				token_swap.clone(),
+				target_public_at_bridged_chain,
+				swap_delivery_and_dispatch_fee,
+				bridged_chain_spec_version,
+				bridged_currency_transfer.encode(),
+				bridged_currency_transfer_weight,
+				bridged_currency_transfer_signature,
+			)
+			.into();
+
+			// start tokens swap
+			let source_genesis_hash = *source_client.genesis_hash();
+			let create_swap_signer = source_sign.clone();
+			let swap_created_at = wait_until_transaction_is_finalized::<Source>(
+				source_client
+					.submit_and_watch_signed_extrinsic(
+						accounts.source_account_at_this_chain.clone(),
+						move |_, transaction_nonce| {
+							Bytes(
+								Source::sign_transaction(
+									source_genesis_hash,
+									&create_swap_signer,
+									relay_substrate_client::TransactionEra::immortal(),
+									UnsignedTransaction::new(create_swap_call, transaction_nonce),
+								)
+								.encode(),
+							)
+						},
+					)
+					.await?,
+			)
+			.await?;
+
+			// read state of swap after it has been created
+			let token_swap_hash: H256 = token_swap.using_encoded(blake2_256).into();
+			let token_swap_storage_key = bp_runtime::storage_map_final_key_identity(
+				TOKEN_SWAP_PALLET_NAME,
+				pallet_bridge_token_swap::PENDING_SWAPS_MAP_NAME,
+				token_swap_hash.as_ref(),
+			);
+			match read_token_swap_state(&source_client, swap_created_at, &token_swap_storage_key).await? {
+				Some(bp_token_swap::TokenSwapState::Started) => {
+					log::info!(target: "bridge", "Swap has been successfully started");
+					let intermediate_balances =
+						read_account_balances(&accounts, &source_client, &target_client).await?;
+					log::info!(target: "bridge", "Intermediate balances: {:?}", intermediate_balances);
+				}
+				Some(token_swap_state) => {
+					return Err(anyhow::format_err!(
+						"Fresh token swap has unexpected state: {:?}",
+						token_swap_state,
+					))
+				}
+				None => return Err(anyhow::format_err!("Failed to start token swap")),
+			};
+
+			//
+			// Step 2: message is being relayed to the target chain and dispathed there
+			//
+
+			// wait until message is dispatched at the target chain and dispatch result delivered back to source chain
+			let token_swap_state = wait_until_token_swap_state_is_changed(
+				&source_client,
+				&token_swap_storage_key,
+				bp_token_swap::TokenSwapState::Started,
+			)
+			.await?;
+			let is_transfer_succeeded = match token_swap_state {
+				Some(bp_token_swap::TokenSwapState::Started) => {
+					unreachable!("wait_until_token_swap_state_is_changed only returns if state is not Started; qed",)
+				}
+				None => return Err(anyhow::format_err!("Fresh token swap has disappeared unexpectedly")),
+				Some(bp_token_swap::TokenSwapState::Confirmed) => {
+					log::info!(
+						target: "bridge",
+						"Transfer has been successfully dispatched at the target chain. Swap can be claimed",
+					);
+					true
+				}
+				Some(bp_token_swap::TokenSwapState::Failed) => {
+					log::info!(
+						target: "bridge",
+						"Transfer has been dispatched with an error at the target chain. Swap can be cancelled",
+					);
+					false
+				}
+			};
+
+			// by this time: (1) token swap account has been created and (2) if transfer has been successfully
+			// dispatched, both target chain balances have changed
+			let intermediate_balances = read_account_balances(&accounts, &source_client, &target_client).await?;
+			log::info!(target: "bridge", "Intermediate balances: {:?}", intermediate_balances);
+
+			// transfer has been dispatched, but we may need to wait until block where swap can be claimed/cancelled
+			if let bp_token_swap::TokenSwapType::LockClaimUntilBlock(ref last_available_block_number, _) =
+				token_swap.swap_type
+			{
+				wait_until_swap_unlocked(
+					&source_client,
+					last_available_block_number + BlockNumberOf::<Source>::one(),
+				)
+				.await?;
+			}
+
+			//
+			// Step 3: we may now claim or cancel the swap
+			//
+
+			if is_transfer_succeeded {
+				log::info!(target: "bridge", "Claiming the swap swap");
+
+				// prepare `claim_swap` message that will be sent over the bridge
+				let claim_swap_call: CallOf<Source> = pallet_bridge_token_swap::Call::claim_swap(token_swap).into();
+				let claim_swap_message = bp_message_dispatch::MessagePayload {
+					spec_version: SOURCE_SPEC_VERSION,
+					weight: claim_swap_call.get_dispatch_info().weight,
+					origin: bp_message_dispatch::CallOrigin::SourceAccount(
+						accounts.target_account_at_bridged_chain.clone(),
+					),
+					dispatch_fee_payment: bp_runtime::messages::DispatchFeePayment::AtSourceChain,
+					call: claim_swap_call.encode(),
+				};
+				let claim_swap_delivery_and_dispatch_fee: BalanceOf<Target> =
+					crate::cli::estimate_fee::estimate_message_delivery_and_dispatch_fee(
+						&target_client,
+						ESTIMATE_TARGET_TO_SOURCE_MESSAGE_FEE_METHOD,
+						TARGET_TO_SOURCE_LANE_ID,
+						claim_swap_message.clone(),
+					)
+					.await?;
+				let send_message_call: CallOf<Target> = pallet_bridge_messages::Call::send_message(
+					TARGET_TO_SOURCE_LANE_ID,
+					claim_swap_message,
+					claim_swap_delivery_and_dispatch_fee,
+				)
+				.into();
+
+				// send `claim_swap` message
+				let target_genesis_hash = *target_client.genesis_hash();
+				let _ = wait_until_transaction_is_finalized::<Target>(
+					target_client
+						.submit_and_watch_signed_extrinsic(
+							accounts.target_account_at_bridged_chain.clone(),
+							move |_, transaction_nonce| {
+								Bytes(
+									Target::sign_transaction(
+										target_genesis_hash,
+										&target_sign,
+										relay_substrate_client::TransactionEra::immortal(),
+										UnsignedTransaction::new(send_message_call, transaction_nonce),
+									)
+									.encode(),
+								)
+							},
+						)
+						.await?,
+				)
+				.await?;
+
+				// wait until swap state is updated
+				let token_swap_state = wait_until_token_swap_state_is_changed(
+					&source_client,
+					&token_swap_storage_key,
+					bp_token_swap::TokenSwapState::Confirmed,
+				)
+				.await?;
+				if token_swap_state != None {
+					return Err(anyhow::format_err!(
+						"Confirmed token swap state has been changed to {:?} unexpectedly"
+					));
+				}
+			} else {
+				log::info!(target: "bridge", "Cancelling the swap");
+				let cancel_swap_call: CallOf<Source> =
+					pallet_bridge_token_swap::Call::cancel_swap(token_swap.clone()).into();
+				let _ = wait_until_transaction_is_finalized::<Source>(
+					source_client
+						.submit_and_watch_signed_extrinsic(
+							accounts.source_account_at_this_chain.clone(),
+							move |_, transaction_nonce| {
+								Bytes(
+									Source::sign_transaction(
+										source_genesis_hash,
+										&source_sign,
+										relay_substrate_client::TransactionEra::immortal(),
+										UnsignedTransaction::new(cancel_swap_call, transaction_nonce),
+									)
+									.encode(),
+								)
+							},
+						)
+						.await?,
+				)
+				.await?;
+			}
+
+			// print final balances
+			let final_balances = read_account_balances(&accounts, &source_client, &target_client).await?;
+			log::info!(target: "bridge", "Final account balances: {:?}", final_balances);
+
+			Ok(())
+		})
+	}
+
+	/// Prepare token swap intention.
+	async fn prepare_token_swap<Source: CliChain, Target: CliChain>(
+		&self,
+		source_client: &Client<Source>,
+		source_sign: &Source::KeyPair,
+		target_sign: &Target::KeyPair,
+	) -> anyhow::Result<
+		bp_token_swap::TokenSwap<
+			BlockNumberOf<Source>,
+			BalanceOf<Source>,
+			AccountIdOf<Source>,
+			BalanceOf<Target>,
+			AccountIdOf<Target>,
+		>,
+	>
+	where
+		AccountIdOf<Source>: From<<Source::KeyPair as Pair>::Public>,
+		AccountIdOf<Target>: From<<Target::KeyPair as Pair>::Public>,
+		BalanceOf<Source>: From<u64>,
+		BalanceOf<Target>: From<u64>,
+	{
+		// accounts that are directly controlled by participants
+		let source_account_at_this_chain: AccountIdOf<Source> = source_sign.public().into();
+		let target_account_at_bridged_chain: AccountIdOf<Target> = target_sign.public().into();
+
+		// balances that we're going to swap
+		let source_balance_at_this_chain: BalanceOf<Source> = self.source_balance.cast().into();
+		let target_balance_at_bridged_chain: BalanceOf<Target> = self.target_balance.cast().into();
+
+		// prepare token swap intention
+		Ok(bp_token_swap::TokenSwap {
+			swap_type: self.prepare_token_swap_type(&source_client).await?,
+			source_balance_at_this_chain,
+			source_account_at_this_chain: source_account_at_this_chain.clone(),
+			target_balance_at_bridged_chain,
+			target_account_at_bridged_chain: target_account_at_bridged_chain.clone(),
+		})
+	}
+
+	/// Prepare token swap type.
+	async fn prepare_token_swap_type<Source: Chain>(
+		&self,
+		source_client: &Client<Source>,
+	) -> anyhow::Result<bp_token_swap::TokenSwapType<BlockNumberOf<Source>>> {
+		match self.swap_type {
+			TokenSwapType::NoLock => Ok(bp_token_swap::TokenSwapType::TemporaryTargetAccountAtBridgedChain),
+			TokenSwapType::LockUntilBlock {
+				blocks_before_expire,
+				ref swap_nonce,
+			} => {
+				let blocks_before_expire: BlockNumberOf<Source> = blocks_before_expire.into();
+				let current_source_block_number = *source_client.best_header().await?.number();
+				Ok(bp_token_swap::TokenSwapType::LockClaimUntilBlock(
+					current_source_block_number + blocks_before_expire,
+					swap_nonce.unwrap_or_else(|| {
+						U256::from(random::<u128>())
+							.overflowing_mul(U256::from(random::<u128>()))
+							.0
+					}),
+				))
+			}
+		}
+	}
+}
+
+/// Accounts that are participating in the swap.
+#[derive(Debug)]
+struct TokenSwapAccounts<ThisAccountId, BridgedAccountId> {
+	source_account_at_this_chain: ThisAccountId,
+	source_account_at_bridged_chain: BridgedAccountId,
+	target_account_at_bridged_chain: BridgedAccountId,
+	target_account_at_this_chain: ThisAccountId,
+	swap_account: ThisAccountId,
+}
+
+/// Swap accounts balances.
+#[derive(Debug)]
+struct TokenSwapBalances<ThisBalance, BridgedBalance> {
+	source_account_at_this_chain_balance: Option<ThisBalance>,
+	source_account_at_bridged_chain_balance: Option<BridgedBalance>,
+	target_account_at_bridged_chain_balance: Option<BridgedBalance>,
+	target_account_at_this_chain_balance: Option<ThisBalance>,
+	swap_account_balance: Option<ThisBalance>,
+}
+
+/// Read swap accounts balances.
+async fn read_account_balances<Source: ChainWithBalances, Target: ChainWithBalances>(
+	accounts: &TokenSwapAccounts<AccountIdOf<Source>, AccountIdOf<Target>>,
+	source_client: &Client<Source>,
+	target_client: &Client<Target>,
+) -> anyhow::Result<TokenSwapBalances<BalanceOf<Source>, BalanceOf<Target>>> {
+	Ok(TokenSwapBalances {
+		source_account_at_this_chain_balance: read_account_balance(
+			&source_client,
+			&accounts.source_account_at_this_chain,
+		)
+		.await?,
+		source_account_at_bridged_chain_balance: read_account_balance(
+			&target_client,
+			&accounts.source_account_at_bridged_chain,
+		)
+		.await?,
+		target_account_at_bridged_chain_balance: read_account_balance(
+			&target_client,
+			&accounts.target_account_at_bridged_chain,
+		)
+		.await?,
+		target_account_at_this_chain_balance: read_account_balance(
+			&source_client,
+			&accounts.target_account_at_this_chain,
+		)
+		.await?,
+		swap_account_balance: read_account_balance(&source_client, &accounts.swap_account).await?,
+	})
+}
+
+/// Read account balance.
+async fn read_account_balance<C: ChainWithBalances>(
+	client: &Client<C>,
+	account: &AccountIdOf<C>,
+) -> anyhow::Result<Option<BalanceOf<C>>> {
+	match client.free_native_balance(account.clone()).await {
+		Ok(balance) => Ok(Some(balance)),
+		Err(SubstrateError::AccountDoesNotExist) => Ok(None),
+		Err(error) => Err(anyhow::format_err!(
+			"Failed to read balance of {} account {:?}: {:?}",
+			C::NAME,
+			account,
+			error,
+		)),
+	}
+}
+
+/// Wait until transaction is included into finalized block.
+///
+/// Returns the hash of the finalized block with transaction.
+async fn wait_until_transaction_is_finalized<C: Chain>(
+	subscription: Subscription<TransactionStatusOf<C>>,
+) -> anyhow::Result<HashOf<C>> {
+	loop {
+		let transaction_status = subscription.next().await?;
+		match transaction_status {
+			Some(TransactionStatusOf::<C>::FinalityTimeout(_))
+			| Some(TransactionStatusOf::<C>::Usurped(_))
+			| Some(TransactionStatusOf::<C>::Dropped)
+			| Some(TransactionStatusOf::<C>::Invalid)
+			| None => {
+				return Err(anyhow::format_err!(
+					"We've been waiting for finalization of {} transaction, but it now has the {:?} status",
+					C::NAME,
+					transaction_status,
+				))
+			}
+			Some(TransactionStatusOf::<C>::Finalized(block_hash)) => {
+				log::trace!(
+					target: "bridge",
+					"{} transaction has been finalized at block {}",
+					C::NAME,
+					block_hash,
+				);
+				return Ok(block_hash);
+			}
+			_ => {
+				log::trace!(
+					target: "bridge",
+					"Received intermediate status of {} transaction: {:?}",
+					C::NAME,
+					transaction_status,
+				);
+			}
+		}
+	}
+}
+
+/// Waits until token swap state is changed from `Started` to something else.
+async fn wait_until_token_swap_state_is_changed<C: Chain>(
+	client: &Client<C>,
+	swap_state_storage_key: &StorageKey,
+	previous_token_swap_state: bp_token_swap::TokenSwapState,
+) -> anyhow::Result<Option<bp_token_swap::TokenSwapState>> {
+	log::trace!(target: "bridge", "Waiting for token swap state change");
+	loop {
+		async_std::task::sleep(C::AVERAGE_BLOCK_INTERVAL).await;
+
+		let best_block = client.best_finalized_header_number().await?;
+		let best_block_hash = client.block_hash_by_number(best_block).await?;
+		log::trace!(target: "bridge", "Inspecting {} block {}/{}", C::NAME, best_block, best_block_hash);
+
+		let token_swap_state = read_token_swap_state(client, best_block_hash, swap_state_storage_key).await?;
+		match token_swap_state {
+			Some(new_token_swap_state) if new_token_swap_state == previous_token_swap_state => {}
+			_ => {
+				log::trace!(
+					target: "bridge",
+					"Token swap state has been changed from {:?} to {:?}",
+					previous_token_swap_state,
+					token_swap_state,
+				);
+				return Ok(token_swap_state);
+			}
+		}
+	}
+}
+
+/// Waits until swap can be claimed or cancelled.
+async fn wait_until_swap_unlocked<C: Chain>(
+	client: &Client<C>,
+	required_block_number: BlockNumberOf<C>,
+) -> anyhow::Result<()> {
+	log::trace!(target: "bridge", "Waiting for token swap unlock");
+	loop {
+		async_std::task::sleep(C::AVERAGE_BLOCK_INTERVAL).await;
+
+		let best_block = client.best_finalized_header_number().await?;
+		let best_block_hash = client.block_hash_by_number(best_block).await?;
+		if best_block >= required_block_number {
+			return Ok(());
+		}
+
+		log::trace!(target: "bridge", "Skipping {} block {}/{}", C::NAME, best_block, best_block_hash);
+	}
+}
+
+/// Read state of the active token swap.
+async fn read_token_swap_state<C: Chain>(
+	client: &Client<C>,
+	at_block: C::Hash,
+	swap_state_storage_key: &StorageKey,
+) -> anyhow::Result<Option<bp_token_swap::TokenSwapState>> {
+	Ok(client
+		.storage_value(swap_state_storage_key.clone(), Some(at_block))
+		.await?)
+}

bridges/relays/client-substrate/Cargo.toml

@@ -10,6 +10,7 @@ async-std = { version = "1.6.5", features = ["attributes"] }
 async-trait = "0.1.40"
 codec = { package = "parity-scale-codec", version = "2.2.0" }
 jsonrpsee-proc-macros = "0.2"
+jsonrpsee-types = "0.2"
 jsonrpsee-ws-client = "0.2"
 log = "0.4.11"
 num-traits = "0.2"
@@ -37,6 +38,7 @@ sp-finality-grandpa = { git = "https://github.com/paritytech/substrate", branch
 sp-rpc = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-storage = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-transaction-pool = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-trie = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-version = { git = "https://github.com/paritytech/substrate", branch = "master" }
 

bridges/relays/client-substrate/src/chain.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
-use bp_runtime::{Chain as ChainBase, TransactionEraOf};
+use bp_runtime::{Chain as ChainBase, HashOf, TransactionEraOf};
 use codec::{Codec, Encode};
 use frame_support::weights::WeightToFeePolynomial;
 use jsonrpsee_ws_client::{DeserializeOwned, Serialize};
@@ -25,6 +25,7 @@ use sp_runtime::{
 	traits::{Block as BlockT, Dispatchable, Member},
 	EncodedJustification,
 };
+use sp_transaction_pool::TransactionStatus;
 use std::{fmt::Debug, time::Duration};
 
 /// Substrate-based chain from minimal relay-client point of view.
@@ -50,8 +51,12 @@ pub trait Chain: ChainBase + Clone {
 	type WeightToFee: WeightToFeePolynomial<Balance = Self::Balance>;
 }
 
-/// Weight-to-Fee type used by the chain
+/// Call type used by the chain.
+pub type CallOf<C> = <C as Chain>::Call;
+/// Weight-to-Fee type used by the chain.
 pub type WeightToFeeOf<C> = <C as Chain>::WeightToFee;
+/// Transaction status of the chain.
+pub type TransactionStatusOf<C> = TransactionStatus<HashOf<C>, HashOf<C>>;
 
 /// Substrate-based chain with `frame_system::Config::AccountData` set to
 /// the `pallet_balances::AccountData<Balance>`.

bridges/relays/client-substrate/src/client.rs

@@ -16,7 +16,7 @@
 
 //! Substrate node client.
 
-use crate::chain::{Chain, ChainWithBalances};
+use crate::chain::{Chain, ChainWithBalances, TransactionStatusOf};
 use crate::rpc::Substrate;
 use crate::{ConnectionParams, Error, HeaderIdOf, Result};
 
@@ -31,7 +31,7 @@ use num_traits::{Bounded, Zero};
 use pallet_balances::AccountData;
 use pallet_transaction_payment::InclusionFee;
 use relay_utils::{relay_loop::RECONNECT_DELAY, HeaderId};
-use sp_core::{storage::StorageKey, Bytes};
+use sp_core::{storage::StorageKey, Bytes, Hasher};
 use sp_runtime::{
 	traits::Header as HeaderT,
 	transaction_validity::{TransactionSource, TransactionValidity},
@@ -45,7 +45,7 @@ const SUB_API_TXPOOL_VALIDATE_TRANSACTION: &str = "TaggedTransactionQueue_valida
 const MAX_SUBSCRIPTION_CAPACITY: usize = 4096;
 
 /// Opaque justifications subscription type.
-pub struct JustificationsSubscription(Mutex<futures::channel::mpsc::Receiver<Option<Bytes>>>);
+pub struct Subscription<T>(Mutex<futures::channel::mpsc::Receiver<Option<T>>>);
 
 /// Opaque GRANDPA authorities set.
 pub type OpaqueGrandpaAuthoritiesSet = Vec<u8>;
@@ -190,6 +190,14 @@ impl<C: Chain> Client<C> {
 			.await
 	}
 
+	/// Return number of the best finalized block.
+	pub async fn best_finalized_header_number(&self) -> Result<C::BlockNumber> {
+		Ok(*self
+			.header_by_hash(self.best_finalized_header_hash().await?)
+			.await?
+			.number())
+	}
+
 	/// Returns the best Substrate header.
 	pub async fn best_header(&self) -> Result<C::Header>
 	where
@@ -243,9 +251,13 @@ impl<C: Chain> Client<C> {
 	}
 
 	/// Read value from runtime storage.
-	pub async fn storage_value<T: Send + Decode + 'static>(&self, storage_key: StorageKey) -> Result<Option<T>> {
+	pub async fn storage_value<T: Send + Decode + 'static>(
+		&self,
+		storage_key: StorageKey,
+		block_hash: Option<C::Hash>,
+	) -> Result<Option<T>> {
 		self.jsonrpsee_execute(move |client| async move {
-			Substrate::<C>::state_get_storage(&*client, storage_key)
+			Substrate::<C>::state_get_storage(&*client, storage_key, block_hash)
 				.await?
 				.map(|encoded_value| T::decode(&mut &encoded_value.0[..]).map_err(Error::ResponseParseFailed))
 				.transpose()
@@ -260,7 +272,7 @@ impl<C: Chain> Client<C> {
 	{
 		self.jsonrpsee_execute(move |client| async move {
 			let storage_key = C::account_info_storage_key(&account);
-			let encoded_account_data = Substrate::<C>::state_get_storage(&*client, storage_key)
+			let encoded_account_data = Substrate::<C>::state_get_storage(&*client, storage_key, None)
 				.await?
 				.ok_or(Error::AccountDoesNotExist)?;
 			let decoded_account_data =
@@ -318,6 +330,44 @@ impl<C: Chain> Client<C> {
 		.await
 	}
 
+	/// Does exactly the same as `submit_signed_extrinsic`, but keeps watching for extrinsic status
+	/// after submission.
+	pub async fn submit_and_watch_signed_extrinsic(
+		&self,
+		extrinsic_signer: C::AccountId,
+		prepare_extrinsic: impl FnOnce(HeaderIdOf<C>, C::Index) -> Bytes + Send + 'static,
+	) -> Result<Subscription<TransactionStatusOf<C>>> {
+		let _guard = self.submit_signed_extrinsic_lock.lock().await;
+		let transaction_nonce = self.next_account_index(extrinsic_signer).await?;
+		let best_header = self.best_header().await?;
+		let best_header_id = HeaderId(*best_header.number(), best_header.hash());
+		let subscription = self
+			.jsonrpsee_execute(move |client| async move {
+				let extrinsic = prepare_extrinsic(best_header_id, transaction_nonce);
+				let tx_hash = C::Hasher::hash(&extrinsic.0);
+				let subscription = client
+					.subscribe(
+						"author_submitAndWatchExtrinsic",
+						JsonRpcParams::Array(vec![
+							jsonrpsee_types::to_json_value(extrinsic).map_err(|e| Error::RpcError(e.into()))?
+						]),
+						"author_unwatchExtrinsic",
+					)
+					.await?;
+				log::trace!(target: "bridge", "Sent transaction to {} node: {:?}", C::NAME, tx_hash);
+				Ok(subscription)
+			})
+			.await?;
+		let (sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);
+		self.tokio.spawn(Subscription::background_worker(
+			C::NAME.into(),
+			"extrinsic".into(),
+			subscription,
+			sender,
+		));
+		Ok(Subscription(Mutex::new(receiver)))
+	}
+
 	/// Returns pending extrinsics from transaction pool.
 	pub async fn pending_extrinsics(&self) -> Result<Vec<Bytes>> {
 		self.jsonrpsee_execute(
@@ -405,8 +455,8 @@ impl<C: Chain> Client<C> {
 	}
 
 	/// Return new justifications stream.
-	pub async fn subscribe_justifications(&self) -> Result<JustificationsSubscription> {
-		let mut subscription = self
+	pub async fn subscribe_justifications(&self) -> Result<Subscription<Bytes>> {
+		let subscription = self
 			.jsonrpsee_execute(move |client| async move {
 				Ok(client
 					.subscribe(
@@ -417,38 +467,14 @@ impl<C: Chain> Client<C> {
 					.await?)
 			})
 			.await?;
-		let (mut sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);
-		self.tokio.spawn(async move {
-			loop {
-				match subscription.next().await {
-					Ok(Some(justification)) => {
-						if sender.send(Some(justification)).await.is_err() {
-							break;
-						}
-					}
-					Ok(None) => {
-						log::trace!(
-							target: "bridge",
-							"{} justifications subscription stream has returned None. Stream needs to be restarted.",
-							C::NAME,
-						);
-						let _ = sender.send(None).await;
-						break;
-					}
-					Err(e) => {
-						log::trace!(
-							target: "bridge",
-							"{} justifications subscription stream has returned '{:?}'. Stream needs to be restarted.",
-							C::NAME,
-							e,
-						);
-						let _ = sender.send(None).await;
-						break;
-					}
-				}
-			}
-		});
-		Ok(JustificationsSubscription(Mutex::new(receiver)))
+		let (sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);
+		self.tokio.spawn(Subscription::background_worker(
+			C::NAME.into(),
+			"justification".into(),
+			subscription,
+			sender,
+		));
+		Ok(Subscription(Mutex::new(receiver)))
 	}
 
 	/// Execute jsonrpsee future in tokio context.
@@ -465,11 +491,50 @@ impl<C: Chain> Client<C> {
 	}
 }
 
-impl JustificationsSubscription {
-	/// Return next justification from the subscription.
-	pub async fn next(&self) -> Result<Option<Bytes>> {
+impl<T: DeserializeOwned> Subscription<T> {
+	/// Return next item from the subscription.
+	pub async fn next(&self) -> Result<Option<T>> {
 		let mut receiver = self.0.lock().await;
-		let justification = receiver.next().await;
-		Ok(justification.unwrap_or(None))
+		let item = receiver.next().await;
+		Ok(item.unwrap_or(None))
+	}
+
+	/// Background worker that is executed in tokio context as `jsonrpsee` requires.
+	async fn background_worker(
+		chain_name: String,
+		item_type: String,
+		mut subscription: jsonrpsee_types::Subscription<T>,
+		mut sender: futures::channel::mpsc::Sender<Option<T>>,
+	) {
+		loop {
+			match subscription.next().await {
+				Ok(Some(item)) => {
+					if sender.send(Some(item)).await.is_err() {
+						break;
+					}
+				}
+				Ok(None) => {
+					log::trace!(
+						target: "bridge",
+						"{} {} subscription stream has returned None. Stream needs to be restarted.",
+						chain_name,
+						item_type,
+					);
+					let _ = sender.send(None).await;
+					break;
+				}
+				Err(e) => {
+					log::trace!(
+						target: "bridge",
+						"{} {} subscription stream has returned '{:?}'. Stream needs to be restarted.",
+						chain_name,
+						item_type,
+						e,
+					);
+					let _ = sender.send(None).await;
+					break;
+				}
+			}
+		}
 	}
 }

bridges/relays/client-substrate/src/lib.rs

@@ -32,14 +32,15 @@ pub mod metrics;
 use std::time::Duration;
 
 pub use crate::chain::{
-	BlockWithJustification, Chain, ChainWithBalances, TransactionSignScheme, UnsignedTransaction, WeightToFeeOf,
+	BlockWithJustification, CallOf, Chain, ChainWithBalances, TransactionSignScheme, TransactionStatusOf,
+	UnsignedTransaction, WeightToFeeOf,
 };
-pub use crate::client::{Client, JustificationsSubscription, OpaqueGrandpaAuthoritiesSet};
+pub use crate::client::{Client, OpaqueGrandpaAuthoritiesSet, Subscription};
 pub use crate::error::{Error, Result};
 pub use crate::sync_header::SyncHeader;
 pub use bp_runtime::{
-	AccountIdOf, BalanceOf, BlockNumberOf, Chain as ChainBase, HashOf, HeaderOf, IndexOf, TransactionEra,
-	TransactionEraOf,
+	AccountIdOf, AccountPublicOf, BalanceOf, BlockNumberOf, Chain as ChainBase, HashOf, HeaderOf, IndexOf, SignatureOf,
+	TransactionEra, TransactionEraOf,
 };
 
 /// Header id used by the chain.

bridges/relays/client-substrate/src/metrics/float_storage_value.rs

@@ -79,7 +79,7 @@ where
 	async fn update(&self) {
 		let value = self
 			.client
-			.storage_value::<T>(self.storage_key.clone())
+			.storage_value::<T>(self.storage_key.clone(), None)
 			.await
 			.map(|maybe_storage_value| {
 				maybe_storage_value.or(self.maybe_default_value).map(|storage_value| {

bridges/relays/client-substrate/src/rpc.rs

@@ -48,7 +48,7 @@ jsonrpsee_proc_macros::rpc_client_api! {
 		#[rpc(method = "state_call", positional_params)]
 		fn state_call(method: String, data: Bytes, at_block: Option<C::Hash>) -> Bytes;
 		#[rpc(method = "state_getStorage", positional_params)]
-		fn state_get_storage(key: StorageKey) -> Option<StorageData>;
+		fn state_get_storage(key: StorageKey, at_block: Option<C::Hash>) -> Option<StorageData>;
 		#[rpc(method = "state_getReadProof", positional_params)]
 		fn state_prove_storage(keys: Vec<StorageKey>, hash: Option<C::Hash>) -> ReadProof<C::Hash>;
 		#[rpc(method = "state_getRuntimeVersion", positional_params)]