Push evm transactions into a queue. (#311)

Cargo.toml

@@ -76,6 +76,7 @@ members = [
     "tesseract/fees/prisma-cli",
     "tesseract/telemetry",
     "tesseract/config",
+    "tesseract/fisherman",
 
     # integration tests
     "tesseract/integration-test",
@@ -304,6 +305,7 @@ nexus-runtime = { path = "./parachain/runtimes/nexus", default-features = false
 tesseract-primitives = { path = "tesseract/primitives" }
 tesseract-consensus = { path = "tesseract/consensus" }
 tesseract-messaging = { path = "tesseract/messaging" }
+tesseract-fisherman = { path = "tesseract/fisherman" }
 tesseract-substrate = { path = "tesseract/substrate" }
 tesseract-evm = { path = "tesseract/evm" }
 tesseract = { path = "tesseract/relayer" }

tesseract/evm/src/byzantine.rs

@@ -0,0 +1,49 @@
+use std::sync::Arc;
+
+use ethers::providers::Middleware;
+use ismp::{
+	consensus::{StateMachineHeight, StateMachineId},
+	events::StateMachineUpdated,
+};
+use tesseract_primitives::{ByzantineHandler, IsmpProvider};
+
+use crate::EvmClient;
+
+#[async_trait::async_trait]
+impl ByzantineHandler for EvmClient {
+	async fn check_for_byzantine_attack(
+		&self,
+		counterparty: Arc<dyn IsmpProvider>,
+		event: StateMachineUpdated,
+	) -> Result<(), anyhow::Error> {
+		let height = StateMachineHeight {
+			id: StateMachineId {
+				state_id: self.state_machine,
+				consensus_state_id: self.consensus_state_id,
+			},
+			height: event.latest_height,
+		};
+		let Some(header) = self.client.get_block(event.latest_height).await? else {
+			// If block header is not found veto the state commitment
+			log::info!(
+				"Vetoing State Machine Update for {} on {}",
+				self.state_machine,
+				counterparty.state_machine_id().state_id
+			);
+			counterparty.veto_state_commitment(height).await?;
+			return Ok(())
+		};
+
+		let state_machine_commitment = counterparty.query_state_machine_commitment(height).await?;
+		if header.state_root != state_machine_commitment.state_root {
+			log::info!(
+				"Vetoing State Machine Update for {} on {}",
+				self.state_machine,
+				counterparty.state_machine_id().state_id
+			);
+			counterparty.veto_state_commitment(height).await?;
+		}
+
+		Ok(())
+	}
+}

tesseract/evm/src/lib.rs

@@ -8,7 +8,7 @@ use ethers::{
 	signers::Signer,
 };
 use frame_support::crypto::ecdsa::ECDSAExt;
-use ismp::{consensus::ConsensusStateId, events::Event, host::StateMachine};
+use ismp::{consensus::ConsensusStateId, events::Event, host::StateMachine, messaging::Message};
 
 use evm_common::presets::{
 	REQUEST_COMMITMENTS_SLOT, REQUEST_RECEIPTS_SLOT, RESPONSE_COMMITMENTS_SLOT,
@@ -19,9 +19,14 @@ use ismp_solidity_abi::shared_types::{StateCommitment, StateMachineHeight};
 use serde::{Deserialize, Serialize};
 use sp_core::{bytes::from_hex, keccak_256, Pair, H160};
 use std::{sync::Arc, time::Duration};
-use tesseract_primitives::{IsmpProvider, StateMachineUpdated, StreamError};
+use tesseract_primitives::{
+	queue::{start_pipeline, PipelineQueue},
+	IsmpProvider, StateMachineUpdated, StreamError, TxReceipt,
+};
+use tx::handle_message_submission;
 
 pub mod abi;
+mod byzantine;
 mod gas_oracle;
 pub mod provider;
 
@@ -135,6 +140,8 @@ pub struct EvmClient {
 			Option<tokio::sync::broadcast::Sender<Result<StateMachineUpdated, StreamError>>>,
 		>,
 	>,
+	/// Tx submission pipeline
+	queue: Option<Arc<PipelineQueue<Vec<Message>, anyhow::Result<Vec<TxReceipt>>>>>,
 }
 
 impl EvmClient {
@@ -168,7 +175,7 @@ impl EvmClient {
 		};
 
 		let latest_height = client.get_block_number().await?.as_u64();
-		Ok(Self {
+		let mut partial_client = Self {
 			client,
 			signer,
 			address,
@@ -179,7 +186,16 @@ impl EvmClient {
 			chain_id,
 			client_type: config.client_type.unwrap_or_default(),
 			state_machine_update_sender: Arc::new(tokio::sync::Mutex::new(None)),
-		})
+			queue: None,
+		};
+
+		let partial_client_clone = partial_client.clone();
+		let queue = start_pipeline(move |messages| {
+			let client = partial_client_clone.clone();
+			async move { handle_message_submission(&client, messages).await }
+		});
+		partial_client.queue = Some(Arc::new(queue));
+		Ok(partial_client)
 	}
 
 	pub async fn events(&self, from: u64, to: u64) -> Result<Vec<Event>, anyhow::Error> {
@@ -347,6 +363,7 @@ impl Clone for EvmClient {
 			chain_id: self.chain_id.clone(),
 			client_type: self.client_type.clone(),
 			state_machine_update_sender: self.state_machine_update_sender.clone(),
+			queue: self.queue.clone(),
 		}
 	}
 }

tesseract/evm/src/provider.rs

@@ -1,9 +1,7 @@
 use crate::{
 	abi::{beefy::BeefyConsensusState, EvmHost},
 	gas_oracle::is_orbit_chain,
-	state_comitment_key,
-	tx::submit_messages,
-	EvmClient,
+	state_comitment_key, EvmClient,
 };
 use anyhow::{anyhow, Error};
 use beefy_verifier_primitives::ConsensusState;
@@ -17,7 +15,7 @@ use evm_common::types::EvmStateProof;
 use ismp::{
 	consensus::{ConsensusStateId, StateMachineId},
 	events::{Event, StateCommitmentVetoed},
-	messaging::{hash_request, hash_response, Message, StateCommitmentHeight},
+	messaging::{Message, StateCommitmentHeight},
 };
 use ismp_solidity_abi::evm_host::{PostRequestHandledFilter, PostResponseHandledFilter};
 use pallet_ismp_host_executive::{EvmHostParam, HostParam};
@@ -38,15 +36,14 @@ use futures::{stream::FuturesOrdered, FutureExt};
 use ismp::{
 	consensus::{StateCommitment, StateMachineHeight},
 	host::StateMachine,
-	messaging::{CreateConsensusState, ResponseMessage},
-	router::{Request, RequestResponse},
+	messaging::CreateConsensusState,
 };
 use primitive_types::U256;
 use sp_core::{H160, H256};
 use std::{collections::BTreeMap, sync::Arc, time::Duration};
 use tesseract_primitives::{
-	wait_for_challenge_period, BoxStream, EstimateGasReturnParams, Hasher, IsmpProvider, Query,
-	Signature, StateMachineUpdated, StateProofQueryType, TxReceipt,
+	wait_for_challenge_period, BoxStream, EstimateGasReturnParams, IsmpProvider, Query, Signature,
+	StateMachineUpdated, StateProofQueryType, TxReceipt,
 };
 
 #[async_trait::async_trait]
@@ -722,56 +719,12 @@ impl IsmpProvider for EvmClient {
 	}
 
 	async fn submit(&self, messages: Vec<Message>) -> Result<Vec<TxReceipt>, Error> {
-		let receipts = submit_messages(&self, messages.clone()).await?;
-		let height = self.client.get_block_number().await?.low_u64();
-		let mut results = vec![];
-		for msg in messages {
-			match msg {
-				Message::Request(req_msg) =>
-					for post in req_msg.requests {
-						let req = Request::Post(post);
-						let commitment = hash_request::<Hasher>(&req);
-						if receipts.contains(&commitment) {
-							let tx_receipt = TxReceipt::Request {
-								query: Query {
-									source_chain: req.source_chain(),
-									dest_chain: req.dest_chain(),
-									nonce: req.nonce(),
-									commitment,
-								},
-								height,
-							};
-
-							results.push(tx_receipt);
-						}
-					},
-				Message::Response(ResponseMessage {
-					datagram: RequestResponse::Response(resp),
-					..
-				}) =>
-					for res in resp {
-						let commitment = hash_response::<Hasher>(&res);
-						let request_commitment = hash_request::<Hasher>(&res.request());
-						if receipts.contains(&commitment) {
-							let tx_receipt = TxReceipt::Response {
-								query: Query {
-									source_chain: res.source_chain(),
-									dest_chain: res.dest_chain(),
-									nonce: res.nonce(),
-									commitment,
-								},
-								request_commitment,
-								height,
-							};
-
-							results.push(tx_receipt);
-						}
-					},
-				_ => {},
-			}
-		}
-
-		Ok(results)
+		let queue = self
+			.queue
+			.as_ref()
+			.ok_or_else(|| anyhow!("Trasnsaction submission pipeline was not initialized"))?
+			.clone();
+		queue.send(messages).await?
 	}
 
 	fn request_commitment_full_key(&self, commitment: H256) -> Vec<Vec<u8>> {

tesseract/evm/src/tx.rs

@@ -17,8 +17,8 @@ use ethers::{
 };
 use ismp::{
 	host::StateMachine,
-	messaging::{Message, ResponseMessage},
-	router::{RequestResponse, Response},
+	messaging::{hash_request, hash_response, Message, ResponseMessage},
+	router::{Request, RequestResponse, Response},
 };
 use ismp_solidity_abi::{
 	beefy::StateMachineHeight,
@@ -33,6 +33,7 @@ use pallet_ismp::mmr::{LeafIndexAndPos, Proof as MmrProof};
 use primitive_types::{H160, H256, U256};
 use sp_mmr_primitives::utils::NodesUtils;
 use std::{collections::BTreeSet, sync::Arc, time::Duration};
+use tesseract_primitives::{Hasher, Query, TxReceipt};
 
 use crate::gas_oracle::get_current_gas_cost_in_usd;
 
@@ -392,3 +393,59 @@ pub fn get_chain_gas_limit(state_machine: StateMachine) -> u64 {
 		_ => Default::default(),
 	}
 }
+
+pub async fn handle_message_submission(
+	client: &EvmClient,
+	messages: Vec<Message>,
+) -> Result<Vec<TxReceipt>, anyhow::Error> {
+	let receipts = submit_messages(client, messages.clone()).await?;
+	let height = client.client.get_block_number().await?.low_u64();
+	let mut results = vec![];
+	for msg in messages {
+		match msg {
+			Message::Request(req_msg) =>
+				for post in req_msg.requests {
+					let req = Request::Post(post);
+					let commitment = hash_request::<Hasher>(&req);
+					if receipts.contains(&commitment) {
+						let tx_receipt = TxReceipt::Request {
+							query: Query {
+								source_chain: req.source_chain(),
+								dest_chain: req.dest_chain(),
+								nonce: req.nonce(),
+								commitment,
+							},
+							height,
+						};
+
+						results.push(tx_receipt);
+					}
+				},
+			Message::Response(ResponseMessage {
+				datagram: RequestResponse::Response(resp),
+				..
+			}) =>
+				for res in resp {
+					let commitment = hash_response::<Hasher>(&res);
+					let request_commitment = hash_request::<Hasher>(&res.request());
+					if receipts.contains(&commitment) {
+						let tx_receipt = TxReceipt::Response {
+							query: Query {
+								source_chain: res.source_chain(),
+								dest_chain: res.dest_chain(),
+								nonce: res.nonce(),
+								commitment,
+							},
+							request_commitment,
+							height,
+						};
+
+						results.push(tx_receipt);
+					}
+				},
+			_ => {},
+		}
+	}
+
+	Ok(results)
+}

tesseract/fisherman/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "tesseract-fisherman"
+version = "0.1.0"
+edition = "2021"
+description = "A process that checks for malicious consensus updates"
+authors = ["Polytope Labs <[email protected]>"]
+
+[dependencies]
+anyhow = "1.0.75"
+log = "0.4.17"
+futures = "0.3.28"
+tracing = "0.1.40"
+tokio = { workspace = true, features = ["full"] }
+sp-core = { workspace = true, features = ["full_crypto"] }
+
+ismp = { workspace = true }
+pallet-ismp = { workspace = true }
+
+tesseract-primitives = { workspace = true  }
+sc-service = { workspace = true }