Push cancelled messages to retry queue (#388)

Co-authored-by: Seun Lanlege <[email protected]>

Author: Wizdave97

Cargo.lock

@@ -364,6 +364,11 @@ consensus_state_id = "PARA"
 # This exists for testing and development,
 # misuse can cause the rpc to be overloaded with queries
 # initial_height = 1000
+# (Optional)
+# This provides the relayer with the precision for the fee token on this substrate chain
+# Needed for ensuring correct delivery fee estimation
+# It defaults to 6 which is the decimal for USDC and USDT on Polkadot parachains
+# fee_token_decimals = 6
 ```
 
 It is optional to provide the configuration option for any of the connected chains, The only consequence is your relayer will not deliver requests from the ommitted chain as it has no way of querying the associated fees for requests originating from this chain.

docs/pages/developers/network/relayer.mdx

@@ -123,8 +123,8 @@ pub struct Proof {
 pub fn runtime_error_into_rpc_error(e: impl std::fmt::Display) -> ErrorObjectOwned {
 	ErrorObject::owned(
 		9876, // no real reason for this value
-		"Something wrong",
-		Some(format!("{}", e)),
+		format!("{}", e),
+		None::<String>,
 	)
 }
 

modules/pallets/ismp/rpc/src/lib.rs

@@ -21,7 +21,7 @@ use sp_core::{bytes::from_hex, keccak_256, Pair, H160};
 use std::{sync::Arc, time::Duration};
 use tesseract_primitives::{
 	queue::{start_pipeline, PipelineQueue},
-	IsmpProvider, StateMachineUpdated, StreamError, TxReceipt,
+	IsmpProvider, StateMachineUpdated, StreamError, TxResult,
 };
 use tx::handle_message_submission;
 
@@ -144,7 +144,7 @@ pub struct EvmClient {
 		>,
 	>,
 	/// Tx submission pipeline
-	queue: Option<Arc<PipelineQueue<Vec<Message>, anyhow::Result<Vec<TxReceipt>>>>>,
+	queue: Option<Arc<PipelineQueue<Vec<Message>, anyhow::Result<TxResult>>>>,
 }
 
 impl EvmClient {

tesseract/evm/src/lib.rs

@@ -1,5 +1,5 @@
 use crate::{
-	abi::{beefy::BeefyConsensusState, EvmHost},
+	abi::{beefy::BeefyConsensusState, erc_20::Erc20, EvmHost},
 	gas_oracle::is_orbit_chain,
 	state_comitment_key, EvmClient,
 };
@@ -43,7 +43,7 @@ use sp_core::{H160, H256};
 use std::{collections::BTreeMap, sync::Arc, time::Duration};
 use tesseract_primitives::{
 	wait_for_challenge_period, BoxStream, EstimateGasReturnParams, IsmpProvider, Query, Signature,
-	StateMachineUpdated, StateProofQueryType, TxReceipt,
+	StateMachineUpdated, StateProofQueryType, TxResult,
 };
 
 #[async_trait::async_trait]
@@ -721,7 +721,7 @@ impl IsmpProvider for EvmClient {
 		&self,
 		messages: Vec<Message>,
 		_coprocessor: StateMachine,
-	) -> Result<Vec<TxReceipt>, Error> {
+	) -> Result<TxResult, Error> {
 		let queue = self
 			.queue
 			.as_ref()
@@ -850,6 +850,19 @@ impl IsmpProvider for EvmClient {
 	fn max_concurrent_queries(&self) -> usize {
 		self.config.tracing_batch_size.unwrap_or(10)
 	}
+
+	async fn fee_token_decimals(&self) -> Result<u8, anyhow::Error> {
+		let fee_token = match self.query_host_params(self.state_machine).await? {
+			HostParam::EvmHostParam(params) => params.fee_token,
+			_ => Err(anyhow!("Unexpected host params"))?,
+		};
+
+		let contract = Erc20::new(fee_token, self.client.clone());
+
+		let decimals = contract.decimals().call().await?;
+
+		Ok(decimals)
+	}
 }
 
 pub enum CheckTraceForEventParams {

tesseract/evm/src/provider.rs

@@ -34,7 +34,7 @@ use pallet_ismp::offchain::{LeafIndexAndPos, Proof as MmrProof};
 use polkadot_sdk::sp_mmr_primitives::utils::NodesUtils;
 use primitive_types::{H256, U256};
 use std::{collections::BTreeSet, sync::Arc, time::Duration};
-use tesseract_primitives::{Hasher, Query, TxReceipt};
+use tesseract_primitives::{Hasher, Query, TxReceipt, TxResult};
 
 use crate::gas_oracle::get_current_gas_cost_in_usd;
 
@@ -49,9 +49,10 @@ type SolidityFunctionCall<T> = FunctionCall<
 pub async fn submit_messages(
 	client: &EvmClient,
 	messages: Vec<Message>,
-) -> anyhow::Result<BTreeSet<H256>> {
+) -> anyhow::Result<(BTreeSet<H256>, Vec<Message>)> {
 	let calls = generate_contract_calls(client, messages.clone(), false).await?;
 	let mut events = BTreeSet::new();
+	let mut cancelled: Vec<Message> = vec![];
 	for (index, call) in calls.into_iter().enumerate() {
 		let gas_price = call.tx.gas_price();
 		match call.clone().send().await {
@@ -71,6 +72,11 @@ pub async fn submit_messages(
 					retry,
 				)
 				.await?;
+				if matches!(messages[index], Message::Request(_) | Message::Response(_)) &&
+					evs.is_empty()
+				{
+					cancelled.push(messages[index].clone())
+				}
 				events.extend(evs);
 			},
 			Err(err) => {
@@ -102,7 +108,7 @@ pub async fn submit_messages(
 		log::trace!("Got {} receipts from executing on {:?}", events.len(), client.state_machine);
 	}
 
-	Ok(events)
+	Ok((events, cancelled))
 }
 
 #[async_recursion::async_recursion]
@@ -192,7 +198,8 @@ where
 				}
 			}
 
-			Err(anyhow!("Transaction to {:?} was cancelled!", state_machine_clone))?
+			log::error!("Transaction to {:?} was cancelled!", state_machine_clone);
+			Ok(Default::default())
 		}
 	};
 
@@ -427,8 +434,8 @@ pub fn get_chain_gas_limit(state_machine: StateMachine) -> u64 {
 pub async fn handle_message_submission(
 	client: &EvmClient,
 	messages: Vec<Message>,
-) -> Result<Vec<TxReceipt>, anyhow::Error> {
-	let receipts = submit_messages(client, messages.clone()).await?;
+) -> Result<TxResult, anyhow::Error> {
+	let (receipts, cancelled) = submit_messages(client, messages.clone()).await?;
 	let height = client.client.get_block_number().await?.low_u64();
 	let mut results = vec![];
 	for msg in messages {
@@ -477,5 +484,5 @@ pub async fn handle_message_submission(
 		}
 	}
 
-	Ok(results)
+	Ok(TxResult { receipts: results, unsuccessful: cancelled })
 }

tesseract/evm/src/tx.rs

@@ -369,7 +369,7 @@ impl TransactionPayment {
 	/// Delete any unprofitable message
 	pub async fn delete_unprofitable_messages(
 		&self,
-		unprofitable: Vec<i32>,
+		unprofitable: impl IntoIterator<Item = i32>,
 	) -> Result<(), anyhow::Error> {
 		let actions = unprofitable
 			.into_iter()

tesseract/fees/src/lib.rs

@@ -134,6 +134,7 @@ async fn create_clients(
 		initial_height: None,
 		max_concurent_queries: None,
 		poll_interval: None,
+		fee_token_decimals: None,
 	};
 
 	let chain_b_config = SubstrateConfig {
@@ -148,6 +149,7 @@ async fn create_clients(
 		initial_height: None,
 		max_concurent_queries: None,
 		poll_interval: None,
+		fee_token_decimals: None,
 	};
 
 	// setup state machines

tesseract/integration-test/src/lib.rs

@@ -489,12 +489,16 @@ pub async fn return_successful_queries(
 							return Ok((None, None))
 						};
 
-						let mut fee_metadata: Cost = match msg {
-							Message::Request(_) => og_source.query_request_fee_metadata(query.commitment).await?.into(),
-							Message::Response(_) => og_source.query_response_fee_metadata(query.commitment).await?.into(),
+						let fee_metadata = match msg {
+							Message::Request(_) => og_source.query_request_fee_metadata(query.commitment).await?,
+							Message::Response(_) => og_source.query_response_fee_metadata(query.commitment).await?,
 							_ => Err(anyhow!("Unexpected message: {msg:?}"))?
 						};
 
+						// normalize fee_metadata to 18 decimals since gas cost is calculated in 18 decimals
+						let fee_token_decimal = og_source.fee_token_decimals().await?;
+						let mut fee_metadata: Cost = (fee_metadata * U256::from(10u128.pow(18u32.saturating_sub(fee_token_decimal.into()) as u32))).into();
+
 						let profit = (U256::from(minimum_profit_percentage) *
 							total_gas_to_be_expended_in_usd.0) /
 							U256::from(100);

tesseract/messaging/src/events.rs

@@ -35,7 +35,7 @@ use ismp::{consensus::StateMachineHeight, events::Event, host::StateMachine, rou
 
 use tesseract_primitives::{
 	config::RelayerConfig, observe_challenge_period, wait_for_state_machine_update,
-	HandleGetResponse, HyperbridgeClaim, IsmpProvider, StateMachineUpdated, TxReceipt,
+	HandleGetResponse, HyperbridgeClaim, IsmpProvider, StateMachineUpdated, TxReceipt, TxResult,
 };
 use transaction_fees::TransactionPayment;
 
@@ -159,9 +159,13 @@ where
 						hyperbridge,
 						client_map,
 						tx_payment,
-						config,
+						config.clone(),
 						coprocessor,
-						sender,
+						if !config.disable_fee_accumulation.unwrap_or_default() {
+							Some(sender)
+						} else {
+							None
+						},
 					)
 					.await;
 					tracing::error!("{name} terminated with result {res:?}");
@@ -353,7 +357,7 @@ async fn handle_update(
 
 		let res = chain_a.submit(messages.clone(), coprocessor).await;
 		match res {
-			Ok(receipts) => {
+			Ok(TxResult { receipts, unsuccessful }) => {
 				if let Some(sender) = fee_acc_sender {
 					// We should not store messages when they are delivered to hyperbridge
 					if chain_a.state_machine_id().state_id != coprocessor {
@@ -383,6 +387,25 @@ async fn handle_update(
 						}
 					}
 				}
+
+				if !unsuccessful.is_empty() &&
+					config.unprofitable_retry_frequency.is_some() &&
+					chain_a.state_machine_id().state_id != coprocessor
+				{
+					tracing::error!(target: "tesseract", "Some transactions were cancelled and will be retried");
+					tracing::trace!(target: "tesseract", "Persisting {} cancelled transactions going to {} to the db", unsuccessful.len(), chain_a.name());
+					if let Err(err) = tx_payment
+						.store_unprofitable_messages(
+							unsuccessful,
+							chain_a.state_machine_id().state_id,
+						)
+						.await
+					{
+						tracing::error!(
+						"Encountered an error while cancelled messages inside the database {err:?}"
+					)
+					}
+				}
 			},
 			Err(err) => {
 				tracing::error!("Failed to submit transaction to {}: {err:?}", chain_a.name())
@@ -391,7 +414,10 @@ async fn handle_update(
 	}
 
 	// Store currently unprofitable in messages in db
-	if !unprofitable.is_empty() && config.unprofitable_retry_frequency.is_some() {
+	if !unprofitable.is_empty() &&
+		config.unprofitable_retry_frequency.is_some() &&
+		chain_a.state_machine_id().state_id != coprocessor
+	{
 		tracing::trace!(target: "tesseract", "Persisting {} unprofitable messages going to {} to the db", unprofitable.len(), chain_a.name());
 		if let Err(err) = tx_payment
 			.store_unprofitable_messages(unprofitable, chain_a.state_machine_id().state_id)