Refactor indexing

rust/Cargo.toml

@@ -99,6 +99,7 @@ num = "0.4"
 num-bigint = "0.4"
 num-derive = "0.4.0"
 num-traits = "0.2"
+once_cell = "1.18.0"
 parking_lot = "0.12"
 paste = "1.0"
 pretty_env_logger = "0.5.0"

rust/agents/relayer/src/msg/metadata/multisig/merkle_root_multisig.rs

@@ -36,13 +36,11 @@ impl MultisigIsmMetadataBuilder for MerkleRootMultisigMetadataBuilder {
         checkpoint_syncer: &MultisigCheckpointSyncer,
     ) -> Result<Option<MultisigMetadata>> {
         const CTX: &str = "When fetching MerkleRootMultisig metadata";
-        unwrap_or_none_result!(
-            highest_leaf_index,
+        let highest_leaf_index = unwrap_or_none_result!(
             self.highest_known_leaf_index().await,
             debug!("Couldn't get highest known leaf index")
         );
-        unwrap_or_none_result!(
-            leaf_index,
+        let leaf_index = unwrap_or_none_result!(
             self.get_merkle_leaf_id_by_message_id(message.id())
                 .await
                 .context(CTX)?,
@@ -51,8 +49,7 @@ impl MultisigIsmMetadataBuilder for MerkleRootMultisigMetadataBuilder {
                 "No merkle leaf found for message id, must have not been enqueued in the tree"
             )
         );
-        unwrap_or_none_result!(
-            quorum_checkpoint,
+        let quorum_checkpoint = unwrap_or_none_result!(
             checkpoint_syncer
                 .fetch_checkpoint_in_range(
                     validators,
@@ -67,8 +64,7 @@ impl MultisigIsmMetadataBuilder for MerkleRootMultisigMetadataBuilder {
                 highest_leaf_index, "Couldn't get checkpoint in range"
             )
         );
-        unwrap_or_none_result!(
-            proof,
+        let proof = unwrap_or_none_result!(
             self.get_proof(leaf_index, quorum_checkpoint.checkpoint.checkpoint)
                 .await
                 .context(CTX)?,

rust/agents/relayer/src/msg/metadata/multisig/message_id_multisig.rs

@@ -37,8 +37,7 @@ impl MultisigIsmMetadataBuilder for MessageIdMultisigMetadataBuilder {
         let message_id = message.id();
 
         const CTX: &str = "When fetching MessageIdMultisig metadata";
-        unwrap_or_none_result!(
-            leaf_index,
+        let leaf_index = unwrap_or_none_result!(
             self.get_merkle_leaf_id_by_message_id(message_id)
                 .await
                 .context(CTX)?,
@@ -47,8 +46,7 @@ impl MultisigIsmMetadataBuilder for MessageIdMultisigMetadataBuilder {
                 "No merkle leaf found for message id, must have not been enqueued in the tree"
             )
         );
-        unwrap_or_none_result!(
-            quorum_checkpoint,
+        let quorum_checkpoint = unwrap_or_none_result!(
             checkpoint_syncer
                 .fetch_checkpoint(validators, threshold as usize, leaf_index)
                 .await

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

@@ -382,13 +382,12 @@ impl HyperlaneSequenceIndexerStore<HyperlaneMessage> for HyperlaneSqlDb {
 
     /// Gets the block number at which the log occurred.
     async fn retrieve_log_block_number(&self, sequence: u32) -> Result<Option<u64>> {
-        unwrap_or_none_result!(
-            tx_id,
+        let tx_id = unwrap_or_none_result!(
             self.db
                 .retrieve_dispatched_tx_id(self.domain().id(), &self.mailbox_address, sequence)
                 .await?
         );
-        unwrap_or_none_result!(block_id, self.db.retrieve_block_id(tx_id).await?);
+        let block_id = unwrap_or_none_result!(self.db.retrieve_block_id(tx_id).await?);
         Ok(self.db.retrieve_block_number(block_id).await?)
     }
 }

rust/chains/hyperlane-cosmos/Cargo.toml

@@ -19,6 +19,7 @@ url = { workspace = true }
 bech32 = { workspace = true }
 base64 = { workspace = true }
 serde_json = { workspace = true }
+once_cell = { workspace = true }
 tonic = { workspace = true }
 sha2 = { workspace = true }
 ripemd = { workspace = true }

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

@@ -1,18 +1,21 @@
 use async_trait::async_trait;
-use base64::engine::general_purpose::STANDARD;
-use base64::Engine;
+use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
 use cosmrs::tendermint::abci::EventAttribute;
 use hyperlane_core::{
-    ChainResult, ContractLocator, HyperlaneChain, HyperlaneContract, Indexer,
-    InterchainGasPaymaster, SequenceIndexer, U256,
+    ChainCommunicationError, ChainResult, ContractLocator, HyperlaneChain, HyperlaneContract,
+    HyperlaneDomain, HyperlaneProvider, Indexer, InterchainGasPaymaster, InterchainGasPayment,
+    LogMeta, SequenceIndexer, H256, U256,
 };
-use hyperlane_core::{HyperlaneDomain, HyperlaneProvider, InterchainGasPayment, LogMeta, H256};
+use once_cell::sync::Lazy;
 use std::ops::RangeInclusive;
 
-use crate::grpc::WasmGrpcProvider;
-use crate::rpc::{CosmosWasmIndexer, WasmIndexer};
-use crate::signers::Signer;
-use crate::{ConnectionConf, CosmosProvider};
+use crate::{
+    grpc::WasmGrpcProvider,
+    rpc::{CosmosWasmIndexer, ParsedEvent, WasmIndexer},
+    signers::Signer,
+    utils::{CONTRACT_ADDRESS_ATTRIBUTE_KEY, CONTRACT_ADDRESS_ATTRIBUTE_KEY_BASE64},
+    ConnectionConf, CosmosProvider,
+};
 
 /// A reference to a InterchainGasPaymaster contract on some Cosmos chain
 #[derive(Debug)]
@@ -51,69 +54,113 @@ impl CosmosInterchainGasPaymaster {
     }
 }
 
+// ------------------ Indexer ------------------
+
+const MESSAGE_ID_ATTRIBUTE_KEY: &str = "message_id";
+static MESSAGE_ID_ATTRIBUTE_KEY_BASE64: Lazy<String> =
+    Lazy::new(|| BASE64.encode(MESSAGE_ID_ATTRIBUTE_KEY));
+
+const PAYMENT_ATTRIBUTE_KEY: &str = "payment";
+static PAYMENT_ATTRIBUTE_KEY_BASE64: Lazy<String> =
+    Lazy::new(|| BASE64.encode(PAYMENT_ATTRIBUTE_KEY));
+
+const GAS_AMOUNT_ATTRIBUTE_KEY: &str = "gas_amount";
+static GAS_AMOUNT_ATTRIBUTE_KEY_BASE64: Lazy<String> =
+    Lazy::new(|| BASE64.encode(GAS_AMOUNT_ATTRIBUTE_KEY));
+
+const DESTINATION_ATTRIBUTE_KEY: &str = "dest_domain";
+static DESTINATION_ATTRIBUTE_KEY_BASE64: Lazy<String> =
+    Lazy::new(|| BASE64.encode(DESTINATION_ATTRIBUTE_KEY));
+
 /// A reference to a InterchainGasPaymasterIndexer contract on some Cosmos chain
 #[derive(Debug)]
 pub struct CosmosInterchainGasPaymasterIndexer {
     indexer: Box<CosmosWasmIndexer>,
 }
 
 impl CosmosInterchainGasPaymasterIndexer {
+    /// The interchain gas payment event type from the CW contract.
+    const INTERCHAIN_GAS_PAYMENT_EVENT_TYPE: &str = "igp-core-pay-for-gas";
+
     /// create new Cosmos InterchainGasPaymasterIndexer agent
     pub fn new(
         conf: ConnectionConf,
         locator: ContractLocator,
-        event_type: String,
         reorg_period: u32,
-    ) -> Self {
-        let indexer: CosmosWasmIndexer =
-            CosmosWasmIndexer::new(conf, locator, event_type.clone(), reorg_period);
+    ) -> ChainResult<Self> {
+        let indexer = CosmosWasmIndexer::new(
+            conf,
+            locator,
+            Self::INTERCHAIN_GAS_PAYMENT_EVENT_TYPE.into(),
+            reorg_period,
+        )?;
 
-        Self {
+        Ok(Self {
             indexer: Box::new(indexer),
-        }
+        })
     }
 
-    fn get_parser(
-        &self,
-    ) -> fn(attrs: Vec<EventAttribute>) -> ChainResult<Option<InterchainGasPayment>> {
-        |attrs: Vec<EventAttribute>| -> ChainResult<Option<InterchainGasPayment>> {
-            let mut res = InterchainGasPayment::default();
-            for attr in attrs {
-                let key = attr.key.as_str();
-                let value = attr.value;
-                let value = value.as_str();
-
-                match key {
-                    "message_id" => {
-                        res.message_id = H256::from_slice(hex::decode(value)?.as_slice())
-                    }
-                    "bWVzc2FnZV9pZA==" => {
-                        res.message_id = H256::from_slice(
-                            hex::decode(String::from_utf8(STANDARD.decode(value)?)?)?.as_slice(),
-                        )
-                    }
-                    "payment" => res.payment = value.parse()?,
-                    "cGF5bWVudA==" => {
-                        let dec_str = String::from_utf8(STANDARD.decode(value)?)?;
-                        // U256's from_str assumes a radix of 16, so we explicitly use from_dec_str.
-                        res.payment = U256::from_dec_str(dec_str.as_str())?;
-                    }
-                    "gas_amount" => res.gas_amount = value.parse()?,
-                    "Z2FzX2Ftb3VudA==" => {
-                        let dec_str = String::from_utf8(STANDARD.decode(value)?)?;
-                        // U256's from_str assumes a radix of 16, so we explicitly use from_dec_str.
-                        res.gas_amount = U256::from_dec_str(dec_str.as_str())?;
-                    }
-                    "dest_domain" => res.destination = value.parse()?,
-                    "ZGVzdF9kb21haW4=" => {
-                        res.destination = String::from_utf8(STANDARD.decode(value)?)?.parse()?
-                    }
-                    _ => {}
+    fn interchain_gas_payment_parser(
+        attrs: &Vec<EventAttribute>,
+    ) -> ChainResult<ParsedEvent<InterchainGasPayment>> {
+        let mut contract_address: Option<String> = None;
+        let mut gas_payment = IncompleteInterchainGasPayment::default();
+
+        for attr in attrs {
+            let key = attr.key.as_str();
+            let value = attr.value.as_str();
+
+            match key {
+                CONTRACT_ADDRESS_ATTRIBUTE_KEY => {
+                    contract_address = Some(value.to_string());
+                }
+                v if *CONTRACT_ADDRESS_ATTRIBUTE_KEY_BASE64 == v => {
+                    contract_address = Some(String::from_utf8(BASE64.decode(value)?)?);
+                }
+
+                MESSAGE_ID_ATTRIBUTE_KEY => {
+                    gas_payment.message_id = Some(H256::from_slice(hex::decode(value)?.as_slice()));
+                }
+                v if *MESSAGE_ID_ATTRIBUTE_KEY_BASE64 == v => {
+                    gas_payment.message_id = Some(H256::from_slice(
+                        hex::decode(String::from_utf8(BASE64.decode(value)?)?)?.as_slice(),
+                    ));
+                }
+
+                PAYMENT_ATTRIBUTE_KEY => {
+                    gas_payment.payment = Some(U256::from_dec_str(value)?);
+                }
+                v if *PAYMENT_ATTRIBUTE_KEY_BASE64 == v => {
+                    let dec_str = String::from_utf8(BASE64.decode(value)?)?;
+                    // U256's from_str assumes a radix of 16, so we explicitly use from_dec_str.
+                    gas_payment.payment = Some(U256::from_dec_str(dec_str.as_str())?);
+                }
+
+                GAS_AMOUNT_ATTRIBUTE_KEY => {
+                    gas_payment.gas_amount = Some(U256::from_dec_str(value)?);
+                }
+                v if *GAS_AMOUNT_ATTRIBUTE_KEY_BASE64 == v => {
+                    let dec_str = String::from_utf8(BASE64.decode(value)?)?;
+                    // U256's from_str assumes a radix of 16, so we explicitly use from_dec_str.
+                    gas_payment.gas_amount = Some(U256::from_dec_str(dec_str.as_str())?);
+                }
+
+                DESTINATION_ATTRIBUTE_KEY => {
+                    gas_payment.destination = Some(value.parse::<u32>()?);
+                }
+                v if *DESTINATION_ATTRIBUTE_KEY_BASE64 == v => {
+                    gas_payment.destination =
+                        Some(String::from_utf8(BASE64.decode(value)?)?.parse()?);
                 }
-            }
 
-            Ok(Some(res))
+                _ => {}
+            }
         }
+
+        let contract_address = contract_address
+            .ok_or_else(|| ChainCommunicationError::from_other_str("missing contract_address"))?;
+
+        Ok(ParsedEvent::new(contract_address, gas_payment.try_into()?))
     }
 }
 
@@ -123,8 +170,10 @@ impl Indexer<InterchainGasPayment> for CosmosInterchainGasPaymasterIndexer {
         &self,
         range: RangeInclusive<u32>,
     ) -> ChainResult<Vec<(InterchainGasPayment, LogMeta)>> {
-        let parser = self.get_parser();
-        let result = self.indexer.get_range_event_logs(range, parser).await?;
+        let result = self
+            .indexer
+            .get_range_event_logs(range, Self::interchain_gas_payment_parser)
+            .await?;
         Ok(result)
     }
 
@@ -141,3 +190,86 @@ impl SequenceIndexer<InterchainGasPayment> for CosmosInterchainGasPaymasterIndex
         Ok((None, tip))
     }
 }
+
+#[derive(Default)]
+struct IncompleteInterchainGasPayment {
+    message_id: Option<H256>,
+    payment: Option<U256>,
+    gas_amount: Option<U256>,
+    destination: Option<u32>,
+}
+
+impl TryInto<InterchainGasPayment> for IncompleteInterchainGasPayment {
+    type Error = ChainCommunicationError;
+
+    fn try_into(self) -> Result<InterchainGasPayment, Self::Error> {
+        let message_id = self
+            .message_id
+            .ok_or_else(|| ChainCommunicationError::from_other_str("missing message_id"))?;
+        let payment = self
+            .payment
+            .ok_or_else(|| ChainCommunicationError::from_other_str("missing payment"))?;
+        let gas_amount = self
+            .gas_amount
+            .ok_or_else(|| ChainCommunicationError::from_other_str("missing gas_amount"))?;
+        let destination = self
+            .destination
+            .ok_or_else(|| ChainCommunicationError::from_other_str("missing destination"))?;
+
+        Ok(InterchainGasPayment {
+            message_id,
+            payment,
+            gas_amount,
+            destination,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use cosmrs::tendermint::abci::EventAttribute;
+    use hyperlane_core::{InterchainGasPayment, H256, U256};
+    use std::str::FromStr;
+
+    use crate::{rpc::ParsedEvent, utils::event_attributes_from_str};
+
+    use super::*;
+
+    #[test]
+    fn test_interchain_gas_payment_parser() {
+        // Examples from https://rpc-kralum.neutron-1.neutron.org/tx_search?query=%22tx.height%20%3E=%204000000%20AND%20tx.height%20%3C=%204100000%20AND%20wasm-igp-core-pay-for-gas._contract_address%20=%20%27neutron12p8wntzra3vpfcqv05scdx5sa3ftaj6gjcmtm7ynkl0e6crtt4ns8cnrmx%27%22&prove=false&page=1&per_page=100
+
+        let expected = ParsedEvent::new(
+            "neutron12p8wntzra3vpfcqv05scdx5sa3ftaj6gjcmtm7ynkl0e6crtt4ns8cnrmx".into(),
+            InterchainGasPayment {
+                message_id: H256::from_str(
+                    "5dcf6120f8adf4f267eb1a122a85c42eae257fbc872671e93929fbf63daed19b",
+                )
+                .unwrap(),
+                payment: U256::from(2),
+                gas_amount: U256::from(25000),
+                destination: 169,
+            },
+        );
+
+        let assert_parsed_event = |attrs: &Vec<EventAttribute>| {
+            let parsed_event =
+                CosmosInterchainGasPaymasterIndexer::interchain_gas_payment_parser(attrs).unwrap();
+
+            assert_eq!(parsed_event, expected);
+        };
+
+        // Non-base64 version
+        let non_base64_attrs = event_attributes_from_str(
+            r#"[{"key":"_contract_address","value":"neutron12p8wntzra3vpfcqv05scdx5sa3ftaj6gjcmtm7ynkl0e6crtt4ns8cnrmx", "index": true},{"key":"dest_domain","value":"169", "index": true},{"key":"gas_amount","value":"25000", "index": true},{"key":"gas_refunded","value":"0", "index": true},{"key":"gas_required","value":"2", "index": true},{"key":"message_id","value":"5dcf6120f8adf4f267eb1a122a85c42eae257fbc872671e93929fbf63daed19b", "index": true},{"key":"payment","value":"2", "index": true},{"key":"sender","value":"neutron1vdazwhwkh9wy6ue66pjpuvrxcrywv2ww956dq6ls2gh0n7t9f5rs2hydt2", "index": true}]"#,
+        );
+        assert_parsed_event(&non_base64_attrs);
+
+        // Base64 version
+        let base64_attrs = event_attributes_from_str(
+            r#"[{"key":"X2NvbnRyYWN0X2FkZHJlc3M=","value":"bmV1dHJvbjEycDh3bnR6cmEzdnBmY3F2MDVzY2R4NXNhM2Z0YWo2Z2pjbXRtN3lua2wwZTZjcnR0NG5zOGNucm14","index":true},{"key":"ZGVzdF9kb21haW4=","value":"MTY5","index":true},{"key":"Z2FzX2Ftb3VudA==","value":"MjUwMDA=","index":true},{"key":"Z2FzX3JlZnVuZGVk","value":"MA==","index":true},{"key":"Z2FzX3JlcXVpcmVk","value":"Mg==","index":true},{"key":"bWVzc2FnZV9pZA==","value":"NWRjZjYxMjBmOGFkZjRmMjY3ZWIxYTEyMmE4NWM0MmVhZTI1N2ZiYzg3MjY3MWU5MzkyOWZiZjYzZGFlZDE5Yg==","index":true},{"key":"cGF5bWVudA==","value":"Mg==","index":true},{"key":"c2VuZGVy","value":"bmV1dHJvbjF2ZGF6d2h3a2g5d3k2dWU2NnBqcHV2cnhjcnl3djJ3dzk1NmRxNmxzMmdoMG43dDlmNXJzMmh5ZHQy","index":true}]"#,
+        );
+
+        assert_parsed_event(&base64_attrs);
+    }
+}