feat: Scrape Sealevel dispatched messages

rust/main/chains/hyperlane-sealevel/src/error.rs

@@ -1,6 +1,7 @@
-use hyperlane_core::ChainCommunicationError;
+use hyperlane_core::{ChainCommunicationError, H512};
 use solana_client::client_error::ClientError;
 use solana_sdk::pubkey::ParsePubkeyError;
+use solana_transaction_status::EncodedTransaction;
 
 /// Errors from the crates specific to the hyperlane-sealevel
 /// implementation.
@@ -17,6 +18,21 @@ pub enum HyperlaneSealevelError {
     /// Decoding error
     #[error("{0}")]
     Decoding(#[from] solana_sdk::bs58::decode::Error),
+    /// No transaction in block error
+    #[error("{0}")]
+    NoTransactions(String),
+    /// Too many transactions of particular content in block
+    #[error("{0}")]
+    TooManyTransactions(String),
+    /// Unsupported transaction encoding
+    #[error("{0:?}")]
+    UnsupportedTransactionEncoding(EncodedTransaction),
+    /// Unsigned transaction
+    #[error("{0}")]
+    UnsignedTransaction(H512),
+    /// Incorrect transaction
+    #[error("received incorrect transaction, expected hash: {0:?}, received hash: {1:?}")]
+    IncorrectTransaction(Box<H512>, Box<H512>),
 }
 
 impl From<HyperlaneSealevelError> for ChainCommunicationError {

rust/main/chains/hyperlane-sealevel/src/lib.rs

@@ -24,4 +24,6 @@ mod multisig_ism;
 mod provider;
 mod rpc;
 mod trait_builder;
+mod transaction;
+mod utils;
 mod validator_announce;

rust/main/chains/hyperlane-sealevel/src/mailbox.rs

@@ -4,9 +4,6 @@ use std::{collections::HashMap, num::NonZeroU64, ops::RangeInclusive, str::FromS
 
 use async_trait::async_trait;
 use borsh::{BorshDeserialize, BorshSerialize};
-use jsonrpc_core::futures_util::TryFutureExt;
-use tracing::{debug, info, instrument, warn};
-
 use hyperlane_core::{
     accumulator::incremental::IncrementalMerkle, BatchItem, ChainCommunicationError,
     ChainCommunicationError::ContractError, ChainResult, Checkpoint, ContractLocator, Decode as _,
@@ -19,14 +16,19 @@ use hyperlane_sealevel_interchain_security_module_interface::{
     InterchainSecurityModuleInstruction, VerifyInstruction,
 };
 use hyperlane_sealevel_mailbox::{
-    accounts::{DispatchedMessageAccount, InboxAccount, OutboxAccount},
+    accounts::{
+        DispatchedMessageAccount, InboxAccount, OutboxAccount, ProcessedMessage,
+        ProcessedMessageAccount,
+    },
+    instruction,
     instruction::InboxProcess,
     mailbox_dispatched_message_pda_seeds, mailbox_inbox_pda_seeds, mailbox_outbox_pda_seeds,
     mailbox_process_authority_pda_seeds, mailbox_processed_message_pda_seeds,
 };
 use hyperlane_sealevel_message_recipient_interface::{
     HandleInstruction, MessageRecipientInstruction,
 };
+use jsonrpc_core::futures_util::TryFutureExt;
 use serializable_account_meta::SimulationReturnData;
 use solana_account_decoder::{UiAccountEncoding, UiDataSliceConfig};
 use solana_client::{
@@ -51,10 +53,14 @@ use solana_sdk::{
 };
 use solana_transaction_status::{
     EncodedConfirmedBlock, EncodedTransaction, EncodedTransactionWithStatusMeta, TransactionStatus,
-    UiInnerInstructions, UiInstruction, UiMessage, UiParsedInstruction, UiReturnDataEncoding,
-    UiTransaction, UiTransactionReturnData, UiTransactionStatusMeta,
+    UiCompiledInstruction, UiInnerInstructions, UiInstruction, UiMessage, UiParsedInstruction,
+    UiReturnDataEncoding, UiTransaction, UiTransactionReturnData, UiTransactionStatusMeta,
 };
+use tracing::{debug, info, instrument, warn};
 
+use crate::error::HyperlaneSealevelError;
+use crate::transaction::search_dispatched_message_transactions;
+use crate::utils::{decode_h256, decode_h512, from_base58};
 use crate::{ConnectionConf, SealevelProvider, SealevelRpcClient};
 
 const SYSTEM_PROGRAM: &str = "11111111111111111111111111111111";
@@ -653,72 +659,20 @@ impl SealevelMailboxIndexer {
         self.rpc().get_block_height().await
     }
 
-    async fn get_message_with_nonce(
+    async fn get_dispatched_message_with_nonce(
         &self,
         nonce: u32,
     ) -> ChainResult<(Indexed<HyperlaneMessage>, LogMeta)> {
-        let target_message_account_bytes = &[
-            &hyperlane_sealevel_mailbox::accounts::DISPATCHED_MESSAGE_DISCRIMINATOR[..],
-            &nonce.to_le_bytes()[..],
-        ]
-        .concat();
-        let target_message_account_bytes = base64::encode(target_message_account_bytes);
-
-        // First, find all accounts with the matching account data.
-        // To keep responses small in case there is ever more than 1
-        // match, we don't request the full account data, and just request
-        // the `unique_message_pubkey` field.
-        let memcmp = RpcFilterType::Memcmp(Memcmp {
-            // Ignore the first byte, which is the `initialized` bool flag.
-            offset: 1,
-            bytes: MemcmpEncodedBytes::Base64(target_message_account_bytes),
-            encoding: None,
-        });
-        let config = RpcProgramAccountsConfig {
-            filters: Some(vec![memcmp]),
-            account_config: RpcAccountInfoConfig {
-                encoding: Some(UiAccountEncoding::Base64),
-                // Don't return any data
-                data_slice: Some(UiDataSliceConfig {
-                    offset: 1 + 8 + 4 + 8, // the offset to get the `unique_message_pubkey` field
-                    length: 32,            // the length of the `unique_message_pubkey` field
-                }),
-                commitment: Some(CommitmentConfig::finalized()),
-                min_context_slot: None,
-            },
-            with_context: Some(false),
-        };
+        let discriminator = hyperlane_sealevel_mailbox::accounts::DISPATCHED_MESSAGE_DISCRIMINATOR;
+        let offset = 1 + 8 + 4 + 8; // the offset to get the `unique_message_pubkey` field
+        let length = 32; // the length of the `unique_message_pubkey` field
         let accounts = self
-            .rpc()
-            .get_program_accounts_with_config(&self.mailbox.program_id, config)
+            .search_accounts_by_discriminator(&discriminator, nonce, offset, length)
             .await?;
 
-        // Now loop through matching accounts and find the one with a valid account pubkey
-        // that proves it's an actual message storage PDA.
-        let mut valid_message_storage_pda_pubkey = Option::<Pubkey>::None;
-
-        for (pubkey, account) in accounts {
-            let unique_message_pubkey = Pubkey::new(&account.data);
-            let (expected_pubkey, _bump) = Pubkey::try_find_program_address(
-                mailbox_dispatched_message_pda_seeds!(unique_message_pubkey),
-                &self.mailbox.program_id,
-            )
-            .ok_or_else(|| {
-                ChainCommunicationError::from_other_str(
-                    "Could not find program address for unique_message_pubkey",
-                )
-            })?;
-            if expected_pubkey == pubkey {
-                valid_message_storage_pda_pubkey = Some(pubkey);
-                break;
-            }
-        }
-
-        let valid_message_storage_pda_pubkey =
-            valid_message_storage_pda_pubkey.ok_or_else(|| {
-                ChainCommunicationError::from_other_str(
-                    "Could not find valid message storage PDA pubkey",
-                )
+        let valid_message_storage_pda_pubkey = self
+            .search_and_validate_account(accounts, |account| {
+                self.dispatched_message_account(&account)
             })?;
 
         // Now that we have the valid message storage PDA pubkey, we can get the full account data.
@@ -733,20 +687,178 @@ impl SealevelMailboxIndexer {
         let hyperlane_message =
             HyperlaneMessage::read_from(&mut &dispatched_message_account.encoded_message[..])?;
 
+        let block = self
+            .mailbox
+            .provider
+            .rpc()
+            .get_block(dispatched_message_account.slot)
+            .await?;
+        let block_hash = decode_h256(&block.blockhash)?;
+
+        let transactions =
+            block.transactions.ok_or(HyperlaneSealevelError::NoTransactions("block which should contain message dispatch transaction does not contain any transaction".to_owned()))?;
+
+        let transaction_hashes = search_dispatched_message_transactions(
+            &self.mailbox.program_id,
+            &valid_message_storage_pda_pubkey,
+            transactions,
+        );
+
+        // We expect to see that there is only one message dispatch transaction
+        if transaction_hashes.len() > 1 {
+            Err(HyperlaneSealevelError::TooManyTransactions("Block contains more than one dispatch message transaction operating on the same dispatch message store PDA".to_owned()))?
+        }
+
+        let (transaction_index, transaction_hash) = transaction_hashes
+            .into_iter()
+            .next()
+            .ok_or(HyperlaneSealevelError::NoTransactions("block which should contain message dispatch transaction does not contain any after filtering".to_owned()))?;
+
         Ok((
             hyperlane_message.into(),
             LogMeta {
                 address: self.mailbox.program_id.to_bytes().into(),
                 block_number: dispatched_message_account.slot,
                 // TODO: get these when building out scraper support.
                 // It's inconvenient to get these :|
+                block_hash,
+                transaction_id: transaction_hash,
+                transaction_index: transaction_index as u64,
+                log_index: U256::zero(),
+            },
+        ))
+    }
+
+    fn dispatched_message_account(&self, account: &Account) -> ChainResult<Pubkey> {
+        let unique_message_pubkey = Pubkey::new(&account.data);
+        let (expected_pubkey, _bump) = Pubkey::try_find_program_address(
+            mailbox_dispatched_message_pda_seeds!(unique_message_pubkey),
+            &self.mailbox.program_id,
+        )
+        .ok_or_else(|| {
+            ChainCommunicationError::from_other_str(
+                "Could not find program address for unique message pubkey",
+            )
+        })?;
+        Ok(expected_pubkey)
+    }
+
+    async fn get_delivered_message_with_nonce(
+        &self,
+        nonce: u32,
+    ) -> ChainResult<(Indexed<H256>, LogMeta)> {
+        let discriminator = hyperlane_sealevel_mailbox::accounts::PROCESSED_MESSAGE_DISCRIMINATOR;
+        let offset = 1 + 8 + 8; // the offset to get the `message_id` field
+        let length = 32;
+        let accounts = self
+            .search_accounts_by_discriminator(&discriminator, nonce, offset, length)
+            .await?;
+
+        debug!(account_len = ?accounts.len(), "Found accounts with processed message discriminator");
+
+        let valid_message_storage_pda_pubkey = self
+            .search_and_validate_account(accounts, |account| {
+                self.delivered_message_account(&account)
+            })?;
+
+        // Now that we have the valid delivered message storage PDA pubkey,
+        // we can get the full account data.
+        let account = self
+            .rpc()
+            .get_account_with_finalized_commitment(&valid_message_storage_pda_pubkey)
+            .await?;
+        let delivered_message_account = ProcessedMessageAccount::fetch(&mut account.data.as_ref())
+            .map_err(ChainCommunicationError::from_other)?
+            .into_inner();
+        let message_id = delivered_message_account.message_id;
+
+        Ok((
+            message_id.into(),
+            LogMeta {
+                address: self.mailbox.program_id.to_bytes().into(),
+                block_number: delivered_message_account.slot,
+                // TODO: get these when building out scraper support.
+                // It's inconvenient to get these :|
                 block_hash: H256::zero(),
                 transaction_id: H512::zero(),
                 transaction_index: 0,
                 log_index: U256::zero(),
             },
         ))
     }
+
+    fn delivered_message_account(&self, account: &Account) -> ChainResult<Pubkey> {
+        let message_id = H256::from_slice(&account.data);
+        let (expected_pubkey, _bump) = Pubkey::try_find_program_address(
+            mailbox_processed_message_pda_seeds!(message_id),
+            &self.mailbox.program_id,
+        )
+        .ok_or_else(|| {
+            ChainCommunicationError::from_other_str("Could not find program address for message id")
+        })?;
+        Ok(expected_pubkey)
+    }
+
+    fn search_and_validate_account<F>(
+        &self,
+        accounts: Vec<(Pubkey, Account)>,
+        message_account: F,
+    ) -> ChainResult<Pubkey>
+    where
+        F: Fn(&Account) -> ChainResult<Pubkey>,
+    {
+        let mut valid_storage_pda_pubkey = Option::<Pubkey>::None;
+
+        for (pubkey, account) in accounts {
+            let expected_pubkey = message_account(&account)?;
+            if expected_pubkey == pubkey {
+                valid_storage_pda_pubkey = Some(pubkey);
+                break;
+            }
+        }
+
+        let valid_storage_pda_pubkey = valid_storage_pda_pubkey.ok_or_else(|| {
+            ChainCommunicationError::from_other_str("Could not find valid storage PDA pubkey")
+        })?;
+        Ok(valid_storage_pda_pubkey)
+    }
+
+    async fn search_accounts_by_discriminator(
+        &self,
+        discriminator: &[u8; 8],
+        nonce: u32,
+        offset: usize,
+        length: usize,
+    ) -> ChainResult<Vec<(Pubkey, Account)>> {
+        let target_message_account_bytes = &[&discriminator[..], &nonce.to_le_bytes()[..]].concat();
+        let target_message_account_bytes = base64::encode(target_message_account_bytes);
+
+        // First, find all accounts with the matching account data.
+        // To keep responses small in case there is ever more than 1
+        // match, we don't request the full account data, and just request
+        // the `unique_message_pubkey` field.
+        let memcmp = RpcFilterType::Memcmp(Memcmp {
+            // Ignore the first byte, which is the `initialized` bool flag.
+            offset: 1,
+            bytes: MemcmpEncodedBytes::Base64(target_message_account_bytes),
+            encoding: None,
+        });
+        let config = RpcProgramAccountsConfig {
+            filters: Some(vec![memcmp]),
+            account_config: RpcAccountInfoConfig {
+                encoding: Some(UiAccountEncoding::Base64),
+                data_slice: Some(UiDataSliceConfig { offset, length }),
+                commitment: Some(CommitmentConfig::finalized()),
+                min_context_slot: None,
+            },
+            with_context: Some(false),
+        };
+        let accounts = self
+            .rpc()
+            .get_program_accounts_with_config(&self.mailbox.program_id, config)
+            .await?;
+        Ok(accounts)
+    }
 }
 
 #[async_trait]
@@ -774,7 +886,7 @@ impl Indexer<HyperlaneMessage> for SealevelMailboxIndexer {
         let message_capacity = range.end().saturating_sub(*range.start());
         let mut messages = Vec::with_capacity(message_capacity as usize);
         for nonce in range {
-            messages.push(self.get_message_with_nonce(nonce).await?);
+            messages.push(self.get_dispatched_message_with_nonce(nonce).await?);
         }
         Ok(messages)
     }
@@ -788,9 +900,19 @@ impl Indexer<HyperlaneMessage> for SealevelMailboxIndexer {
 impl Indexer<H256> for SealevelMailboxIndexer {
     async fn fetch_logs_in_range(
         &self,
-        _range: RangeInclusive<u32>,
+        range: RangeInclusive<u32>,
     ) -> ChainResult<Vec<(Indexed<H256>, LogMeta)>> {
-        todo!()
+        info!(
+            ?range,
+            "Fetching SealevelMailboxIndexer HyperlaneMessage Delivery logs"
+        );
+
+        let message_capacity = range.end().saturating_sub(*range.start());
+        let mut message_ids = Vec::with_capacity(message_capacity as usize);
+        for nonce in range {
+            message_ids.push(self.get_delivered_message_with_nonce(nonce).await?);
+        }
+        Ok(message_ids)
     }
 
     async fn get_finalized_block_number(&self) -> ChainResult<u32> {