Merge branch 'main' into kunal/avs-contracts

.github/workflows/test.yml

@@ -5,7 +5,8 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
+    branches:
+      - '*' # run against all branches
   # Allows you to run this workflow manually from the Actions tab
   workflow_dispatch:
 
@@ -175,6 +176,7 @@ jobs:
 
   e2e-matrix:
     runs-on: larger-runner
+    if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.base_ref == 'main')
     needs: [yarn-build]
     strategy:
       matrix:
@@ -264,6 +266,7 @@ jobs:
 
   cli-e2e:
     runs-on: larger-runner
+    if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.base_ref == 'main')
     needs: [yarn-build]
     strategy:
       matrix:

rust/agents/relayer/Cargo.toml

@@ -31,7 +31,7 @@ serde.workspace = true
 serde_json.workspace = true
 strum.workspace = true
 thiserror.workspace = true
-tokio = { workspace = true, features = ["rt", "macros", "parking_lot"] }
+tokio = { workspace = true, features = ["rt", "macros", "parking_lot", "rt-multi-thread"] }
 tracing-futures.workspace = true
 tracing.workspace = true
 

rust/agents/relayer/src/main.rs

@@ -21,7 +21,7 @@ mod relayer;
 mod server;
 mod settings;
 
-#[tokio::main(flavor = "current_thread")]
+#[tokio::main(flavor = "multi_thread", worker_threads = 20)]
 async fn main() -> Result<()> {
     agent_main::<Relayer>().await
 }

rust/agents/relayer/src/msg/mod.rs

@@ -28,7 +28,7 @@
 pub(crate) mod gas_payment;
 pub(crate) mod metadata;
 pub(crate) mod op_queue;
+pub(crate) mod op_submitter;
 pub(crate) mod pending_message;
 pub(crate) mod pending_operation;
 pub(crate) mod processor;
-pub(crate) mod serial_submitter;

rust/agents/relayer/src/msg/processor.rs

@@ -242,7 +242,7 @@ mod test {
                     url: "http://example.com".parse().unwrap(),
                 },
                 transaction_overrides: Default::default(),
-                message_batch: Default::default(),
+                operation_batch: Default::default(),
             }),
             metrics_conf: Default::default(),
             index: Default::default(),

rust/agents/relayer/src/relayer.rs

@@ -33,9 +33,9 @@ use crate::{
         gas_payment::GasPaymentEnforcer,
         metadata::{BaseMetadataBuilder, IsmAwareAppContextClassifier},
         op_queue::QueueOperation,
+        op_submitter::{SerialSubmitter, SerialSubmitterMetrics},
         pending_message::{MessageContext, MessageSubmissionMetrics},
         processor::{MessageProcessor, MessageProcessorMetrics},
-        serial_submitter::{SerialSubmitter, SerialSubmitterMetrics},
     },
     server::{self as relayer_server, MessageRetryRequest},
     settings::{matching_list::MatchingList, RelayerSettings},
@@ -315,7 +315,7 @@ impl BaseAgent for Relayer {
                     // Default to submitting one message at a time if there is no batch config
                     self.core.settings.chains[dest_domain.name()]
                         .connection
-                        .message_batch_config()
+                        .operation_batch_config()
                         .map(|c| c.max_batch_size)
                         .unwrap_or(1),
                 ),
@@ -365,7 +365,7 @@ impl Relayer {
                 .sync("dispatched_messages", cursor)
                 .await
         })
-        .instrument(info_span!("ContractSync"))
+        .instrument(info_span!("MessageSync"))
     }
 
     async fn run_interchain_gas_payment_sync(
@@ -380,7 +380,7 @@ impl Relayer {
             .clone();
         let cursor = contract_sync.cursor(index_settings).await;
         tokio::spawn(async move { contract_sync.clone().sync("gas_payments", cursor).await })
-            .instrument(info_span!("ContractSync"))
+            .instrument(info_span!("IgpSync"))
     }
 
     async fn run_merkle_tree_hook_syncs(
@@ -391,7 +391,7 @@ impl Relayer {
         let contract_sync = self.merkle_tree_hook_syncs.get(origin).unwrap().clone();
         let cursor = contract_sync.cursor(index_settings).await;
         tokio::spawn(async move { contract_sync.clone().sync("merkle_tree_hook", cursor).await })
-            .instrument(info_span!("ContractSync"))
+            .instrument(info_span!("MerkleTreeHookSync"))
     }
 
     fn run_message_processor(

rust/agents/scraper/src/chain_scraper/mod.rs

@@ -9,7 +9,7 @@ use hyperlane_base::settings::IndexSettings;
 use hyperlane_core::{
     unwrap_or_none_result, BlockInfo, Delivery, HyperlaneDomain, HyperlaneLogStore,
     HyperlaneMessage, HyperlaneProvider, HyperlaneSequenceAwareIndexerStoreReader,
-    HyperlaneWatermarkedLogStore, InterchainGasPayment, LogMeta, H256,
+    HyperlaneWatermarkedLogStore, Indexed, InterchainGasPayment, LogMeta, H256,
 };
 use itertools::Itertools;
 use tracing::trace;
@@ -269,7 +269,7 @@ impl HyperlaneSqlDb {
 #[async_trait]
 impl HyperlaneLogStore<HyperlaneMessage> for HyperlaneSqlDb {
     /// Store messages from the origin mailbox into the database.
-    async fn store_logs(&self, messages: &[(HyperlaneMessage, LogMeta)]) -> Result<u32> {
+    async fn store_logs(&self, messages: &[(Indexed<HyperlaneMessage>, LogMeta)]) -> Result<u32> {
         if messages.is_empty() {
             return Ok(0);
         }
@@ -287,7 +287,7 @@ impl HyperlaneLogStore<HyperlaneMessage> for HyperlaneSqlDb {
                 )
                 .unwrap();
             StorableMessage {
-                msg: m.0.clone(),
+                msg: m.0.inner().clone(),
                 meta: &m.1,
                 txn_id: txn.id,
             }
@@ -302,7 +302,7 @@ impl HyperlaneLogStore<HyperlaneMessage> for HyperlaneSqlDb {
 
 #[async_trait]
 impl HyperlaneLogStore<Delivery> for HyperlaneSqlDb {
-    async fn store_logs(&self, deliveries: &[(Delivery, LogMeta)]) -> Result<u32> {
+    async fn store_logs(&self, deliveries: &[(Indexed<Delivery>, LogMeta)]) -> Result<u32> {
         if deliveries.is_empty() {
             return Ok(0);
         }
@@ -322,7 +322,7 @@ impl HyperlaneLogStore<Delivery> for HyperlaneSqlDb {
                 .unwrap()
                 .id;
             StorableDelivery {
-                message_id: *message_id,
+                message_id: *message_id.inner(),
                 meta,
                 txn_id,
             }
@@ -338,7 +338,10 @@ impl HyperlaneLogStore<Delivery> for HyperlaneSqlDb {
 
 #[async_trait]
 impl HyperlaneLogStore<InterchainGasPayment> for HyperlaneSqlDb {
-    async fn store_logs(&self, payments: &[(InterchainGasPayment, LogMeta)]) -> Result<u32> {
+    async fn store_logs(
+        &self,
+        payments: &[(Indexed<InterchainGasPayment>, LogMeta)],
+    ) -> Result<u32> {
         if payments.is_empty() {
             return Ok(0);
         }
@@ -358,7 +361,7 @@ impl HyperlaneLogStore<InterchainGasPayment> for HyperlaneSqlDb {
                 .unwrap()
                 .id;
             StorablePayment {
-                payment,
+                payment: payment.inner(),
                 meta,
                 txn_id,
             }

rust/chains/hyperlane-cosmos/src/interchain_gas.rs

@@ -3,8 +3,8 @@ use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
 use futures::future;
 use hyperlane_core::{
     ChainCommunicationError, ChainResult, ContractLocator, HyperlaneChain, HyperlaneContract,
-    HyperlaneDomain, HyperlaneProvider, Indexer, InterchainGasPaymaster, InterchainGasPayment,
-    LogMeta, SequenceAwareIndexer, H256, U256,
+    HyperlaneDomain, HyperlaneProvider, Indexed, Indexer, InterchainGasPaymaster,
+    InterchainGasPayment, LogMeta, SequenceAwareIndexer, H256, U256,
 };
 use once_cell::sync::Lazy;
 use std::ops::RangeInclusive;
@@ -205,7 +205,7 @@ impl Indexer<InterchainGasPayment> for CosmosInterchainGasPaymasterIndexer {
     async fn fetch_logs(
         &self,
         range: RangeInclusive<u32>,
-    ) -> ChainResult<Vec<(InterchainGasPayment, LogMeta)>> {
+    ) -> ChainResult<Vec<(Indexed<InterchainGasPayment>, LogMeta)>> {
         let logs_futures: Vec<_> = range
             .map(|block_number| {
                 let self_clone = self.clone();
@@ -239,6 +239,7 @@ impl Indexer<InterchainGasPayment> for CosmosInterchainGasPaymasterIndexer {
             .collect::<Result<Vec<_>, _>>()?
             .into_iter()
             .flatten()
+            .map(|(log, meta)| (Indexed::new(log), meta))
             .collect();
 
         Ok(result)

rust/chains/hyperlane-cosmos/src/mailbox.rs

@@ -25,8 +25,8 @@ use tendermint::abci::EventAttribute;
 use crate::utils::{CONTRACT_ADDRESS_ATTRIBUTE_KEY, CONTRACT_ADDRESS_ATTRIBUTE_KEY_BASE64};
 use hyperlane_core::{
     utils::bytes_to_hex, ChainResult, HyperlaneChain, HyperlaneContract, HyperlaneDomain,
-    HyperlaneMessage, HyperlaneProvider, Indexer, LogMeta, Mailbox, TxCostEstimate, TxOutcome,
-    H256, U256,
+    HyperlaneMessage, HyperlaneProvider, Indexed, Indexer, LogMeta, Mailbox, TxCostEstimate,
+    TxOutcome, H256, U256,
 };
 use hyperlane_core::{
     ChainCommunicationError, ContractLocator, Decode, RawHyperlaneMessage, SequenceAwareIndexer,
@@ -353,7 +353,7 @@ impl Indexer<HyperlaneMessage> for CosmosMailboxIndexer {
     async fn fetch_logs(
         &self,
         range: RangeInclusive<u32>,
-    ) -> ChainResult<Vec<(HyperlaneMessage, LogMeta)>> {
+    ) -> ChainResult<Vec<(Indexed<HyperlaneMessage>, LogMeta)>> {
         let logs_futures: Vec<_> = range
             .map(|block_number| {
                 let self_clone = self.clone();
@@ -384,6 +384,7 @@ impl Indexer<HyperlaneMessage> for CosmosMailboxIndexer {
                 }
             })
             .flatten()
+            .map(|(log, meta)| (log.into(), meta))
             .collect();
 
         Ok(result)
@@ -396,7 +397,10 @@ impl Indexer<HyperlaneMessage> for CosmosMailboxIndexer {
 
 #[async_trait]
 impl Indexer<H256> for CosmosMailboxIndexer {
-    async fn fetch_logs(&self, range: RangeInclusive<u32>) -> ChainResult<Vec<(H256, LogMeta)>> {
+    async fn fetch_logs(
+        &self,
+        range: RangeInclusive<u32>,
+    ) -> ChainResult<Vec<(Indexed<H256>, LogMeta)>> {
         // TODO: implement when implementing Cosmos scraping
         todo!()
     }

rust/chains/hyperlane-cosmos/src/merkle_tree_hook.rs

@@ -6,7 +6,7 @@ use futures::future;
 use hyperlane_core::{
     accumulator::incremental::IncrementalMerkle, ChainCommunicationError, ChainResult, Checkpoint,
     ContractLocator, HyperlaneChain, HyperlaneContract, HyperlaneDomain, HyperlaneProvider,
-    Indexer, LogMeta, MerkleTreeHook, MerkleTreeInsertion, SequenceAwareIndexer, H256,
+    Indexed, Indexer, LogMeta, MerkleTreeHook, MerkleTreeInsertion, SequenceAwareIndexer, H256,
 };
 use once_cell::sync::Lazy;
 use tendermint::abci::EventAttribute;
@@ -286,7 +286,7 @@ impl Indexer<MerkleTreeInsertion> for CosmosMerkleTreeHookIndexer {
     async fn fetch_logs(
         &self,
         range: RangeInclusive<u32>,
-    ) -> ChainResult<Vec<(MerkleTreeInsertion, LogMeta)>> {
+    ) -> ChainResult<Vec<(Indexed<MerkleTreeInsertion>, LogMeta)>> {
         let logs_futures: Vec<_> = range
             .map(|block_number| {
                 let self_clone = self.clone();
@@ -317,6 +317,7 @@ impl Indexer<MerkleTreeInsertion> for CosmosMerkleTreeHookIndexer {
                 }
             })
             .flatten()
+            .map(|(log, meta)| (log.into(), meta))
             .collect();
 
         Ok(result)

rust/chains/hyperlane-cosmos/src/trait_builder.rs

@@ -1,7 +1,7 @@
 use std::str::FromStr;
 
 use derive_new::new;
-use hyperlane_core::{ChainCommunicationError, FixedPointNumber};
+use hyperlane_core::{config::OperationBatchConfig, ChainCommunicationError, FixedPointNumber};
 use url::Url;
 
 /// Cosmos connection configuration
@@ -25,6 +25,8 @@ pub struct ConnectionConf {
     /// Cosmos address lengths are sometimes less than 32 bytes, so this helps to serialize it in
     /// bech32 with the appropriate length.
     contract_address_bytes: usize,
+    /// Operation batching configuration
+    pub operation_batch: OperationBatchConfig,
 }
 
 /// Untyped cosmos amount
@@ -112,6 +114,7 @@ impl ConnectionConf {
     }
 
     /// Create a new connection configuration
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         grpc_urls: Vec<Url>,
         rpc_url: String,
@@ -120,6 +123,7 @@ impl ConnectionConf {
         canonical_asset: String,
         minimum_gas_price: RawCosmosAmount,
         contract_address_bytes: usize,
+        operation_batch: OperationBatchConfig,
     ) -> Self {
         Self {
             grpc_urls,
@@ -129,6 +133,7 @@ impl ConnectionConf {
             canonical_asset,
             gas_price: minimum_gas_price,
             contract_address_bytes,
+            operation_batch,
         }
     }
 }

rust/chains/hyperlane-ethereum/src/config.rs

@@ -1,4 +1,4 @@
-use hyperlane_core::{H256, U256};
+use hyperlane_core::{config::OperationBatchConfig, U256};
 use url::Url;
 
 /// Ethereum RPC connection configuration
@@ -33,8 +33,8 @@ pub struct ConnectionConf {
     pub rpc_connection: RpcConnectionConf,
     /// Transaction overrides to use when sending transactions.
     pub transaction_overrides: TransactionOverrides,
-    /// Message batching configuration
-    pub message_batch: MessageBatchConfig,
+    /// Operation batching configuration
+    pub operation_batch: OperationBatchConfig,
 }
 
 /// Ethereum transaction overrides.
@@ -51,12 +51,3 @@ pub struct TransactionOverrides {
     /// Max priority fee per gas to use for EIP-1559 transactions.
     pub max_priority_fee_per_gas: Option<U256>,
 }
-
-/// Config for batching messages
-#[derive(Debug, Clone, Default)]
-pub struct MessageBatchConfig {
-    /// Optional Multicall3 contract address
-    pub multicall3_address: Option<H256>,
-    /// Batch size
-    pub max_batch_size: u32,
-}

rust/chains/hyperlane-ethereum/src/contracts/interchain_gas.rs

@@ -9,8 +9,8 @@ use async_trait::async_trait;
 use ethers::prelude::Middleware;
 use hyperlane_core::{
     ChainCommunicationError, ChainResult, ContractLocator, HyperlaneAbi, HyperlaneChain,
-    HyperlaneContract, HyperlaneDomain, HyperlaneProvider, Indexer, InterchainGasPaymaster,
-    InterchainGasPayment, LogMeta, SequenceAwareIndexer, H160, H256,
+    HyperlaneContract, HyperlaneDomain, HyperlaneProvider, Indexed, Indexer,
+    InterchainGasPaymaster, InterchainGasPayment, LogMeta, SequenceAwareIndexer, H160, H256,
 };
 use tracing::instrument;
 
@@ -89,7 +89,7 @@ where
     async fn fetch_logs(
         &self,
         range: RangeInclusive<u32>,
-    ) -> ChainResult<Vec<(InterchainGasPayment, LogMeta)>> {
+    ) -> ChainResult<Vec<(Indexed<InterchainGasPayment>, LogMeta)>> {
         let events = self
             .contract
             .gas_payment_filter()
@@ -102,12 +102,12 @@ where
             .into_iter()
             .map(|(log, log_meta)| {
                 (
-                    InterchainGasPayment {
+                    Indexed::new(InterchainGasPayment {
                         message_id: H256::from(log.message_id),
                         destination: log.destination_domain,
                         payment: log.payment.into(),
                         gas_amount: log.gas_amount.into(),
-                    },
+                    }),
                     log_meta.into(),
                 )
             })