Refactor agent event indexing

.github/workflows/e2e.yml

@@ -60,4 +60,4 @@ jobs:
         env:
           E2E_CI_MODE: 'true'
           E2E_CI_TIMEOUT_SEC: '600'
-          E2E_KATHY_ROUNDS: '2'
+          E2E_KATHY_MESSAGES: '20'

package.json

@@ -16,6 +16,7 @@
   "private": true,
   "scripts": {
     "build": "yarn workspaces foreach --verbose --parallel --topological run build",
+    "build:e2e": "yarn workspaces foreach --verbose --parallel --exclude @hyperlane-xyz/monorepo --exclude @hyperlane-xyz/hyperlane-token --topological run build",
     "clean": "yarn workspaces foreach --verbose --parallel run clean",
     "sync-submodules": "./scripts/sync-submodules.sh",
     "postinstall": "husky install",

rust/agents/relayer/src/merkle_tree_builder.rs

@@ -3,7 +3,7 @@ use std::fmt::Display;
 use eyre::Result;
 use tracing::{debug, error, instrument};
 
-use hyperlane_base::db::{DbError, HyperlaneDB};
+use hyperlane_base::db::{DbError, HyperlaneRocksDB};
 use hyperlane_core::{
     accumulator::{incremental::IncrementalMerkle, merkle::Proof},
     ChainCommunicationError, H256,
@@ -14,7 +14,7 @@ use crate::prover::{Prover, ProverError};
 /// Struct to sync prover.
 #[derive(Debug)]
 pub struct MerkleTreeBuilder {
-    db: HyperlaneDB,
+    db: HyperlaneRocksDB,
     prover: Prover,
     incremental: IncrementalMerkle,
 }
@@ -68,7 +68,7 @@ pub enum MerkleTreeBuilderError {
 }
 
 impl MerkleTreeBuilder {
-    pub fn new(db: HyperlaneDB) -> Self {
+    pub fn new(db: HyperlaneRocksDB) -> Self {
         let prover = Prover::default();
         let incremental = IncrementalMerkle::default();
         Self {

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

@@ -4,7 +4,7 @@ use async_trait::async_trait;
 use eyre::Result;
 use tracing::{debug, error, trace};
 
-use hyperlane_base::db::HyperlaneDB;
+use hyperlane_base::db::HyperlaneRocksDB;
 use hyperlane_core::{
     HyperlaneMessage, InterchainGasExpenditure, InterchainGasPayment, TxCostEstimate, TxOutcome,
     U256,
@@ -40,13 +40,13 @@ pub struct GasPaymentEnforcer {
     /// policy or another. If a message matches multiple policies'
     /// whitelists, then whichever is first in the list will be used.
     policies: Vec<(Box<dyn GasPaymentPolicy>, MatchingList)>,
-    db: HyperlaneDB,
+    db: HyperlaneRocksDB,
 }
 
 impl GasPaymentEnforcer {
     pub fn new(
         policy_configs: impl IntoIterator<Item = GasPaymentEnforcementConf>,
-        db: HyperlaneDB,
+        db: HyperlaneRocksDB,
     ) -> Self {
         let policies = policy_configs
             .into_iter()
@@ -136,7 +136,7 @@ impl GasPaymentEnforcer {
 mod test {
     use std::str::FromStr;
 
-    use hyperlane_base::db::{test_utils, HyperlaneDB};
+    use hyperlane_base::db::{test_utils, HyperlaneRocksDB};
     use hyperlane_core::{HyperlaneDomain, HyperlaneMessage, TxCostEstimate, H160, H256, U256};
 
     use crate::settings::{
@@ -148,7 +148,7 @@ mod test {
     #[tokio::test]
     async fn test_empty_whitelist() {
         test_utils::run_test_db(|db| async move {
-            let hyperlane_db = HyperlaneDB::new(
+            let hyperlane_db = HyperlaneRocksDB::new(
                 &HyperlaneDomain::new_test_domain("test_empty_whitelist"),
                 db,
             );
@@ -185,7 +185,7 @@ mod test {
         #[allow(unused_must_use)]
         test_utils::run_test_db(|db| async move {
             let hyperlane_db =
-                HyperlaneDB::new(&HyperlaneDomain::new_test_domain("test_no_match"), db);
+                HyperlaneRocksDB::new(&HyperlaneDomain::new_test_domain("test_no_match"), db);
             let matching_list = serde_json::from_str(r#"[{"originDomain": 234}]"#).unwrap();
             let enforcer = GasPaymentEnforcer::new(
                 // Require a payment
@@ -212,7 +212,7 @@ mod test {
     #[tokio::test]
     async fn test_non_empty_matching_list() {
         test_utils::run_test_db(|db| async move {
-            let hyperlane_db = HyperlaneDB::new(&HyperlaneDomain::new_test_domain("test_non_empty_matching_list"), db);
+            let hyperlane_db = HyperlaneRocksDB::new(&HyperlaneDomain::new_test_domain("test_non_empty_matching_list"), db);
 
             let sender_address = "0xaa000000000000000000000000000000000000aa";
             let recipient_address = "0xbb000000000000000000000000000000000000bb";

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

@@ -5,10 +5,11 @@ use std::time::{Duration, Instant};
 use async_trait::async_trait;
 use derive_new::new;
 use eyre::{Context, Result};
+use hyperlane_base::db::HyperlaneRocksDB;
 use prometheus::{IntCounter, IntGauge};
 use tracing::{debug, error, info, instrument, trace, warn};
 
-use hyperlane_base::{db::HyperlaneDB, CachingMailbox, CoreMetrics};
+use hyperlane_base::CoreMetrics;
 use hyperlane_core::{HyperlaneChain, HyperlaneDomain, HyperlaneMessage, Mailbox, U256};
 
 use super::{
@@ -29,9 +30,9 @@ const CONFIRM_DELAY: Duration = if cfg!(any(test, feature = "test-utils")) {
 /// instance is for a unique origin -> destination pairing.
 pub struct MessageContext {
     /// Mailbox on the destination chain.
-    pub destination_mailbox: CachingMailbox,
+    pub destination_mailbox: Arc<dyn Mailbox>,
     /// Origin chain database to verify gas payments.
-    pub origin_db: HyperlaneDB,
+    pub origin_db: HyperlaneRocksDB,
     /// Used to construct the ISM metadata needed to verify a message from the
     /// origin.
     pub metadata_builder: BaseMetadataBuilder,

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

@@ -10,7 +10,7 @@ use tokio::{
 };
 use tracing::{debug, info_span, instrument, instrument::Instrumented, trace, Instrument};
 
-use hyperlane_base::{db::HyperlaneDB, CoreMetrics};
+use hyperlane_base::{db::HyperlaneRocksDB, CoreMetrics};
 use hyperlane_core::{HyperlaneDomain, HyperlaneMessage};
 
 use crate::msg::pending_operation::DynPendingOperation;
@@ -22,7 +22,7 @@ use super::pending_message::*;
 /// for to the appropriate destination.
 #[derive(new)]
 pub struct MessageProcessor {
-    db: HyperlaneDB,
+    db: HyperlaneRocksDB,
     whitelist: Arc<MatchingList>,
     blacklist: Arc<MatchingList>,
     metrics: MessageProcessorMetrics,
@@ -62,7 +62,7 @@ impl MessageProcessor {
 
     #[instrument(ret, err, skip(self), level = "info")]
     async fn main_loop(mut self) -> Result<()> {
-        // Forever, scan HyperlaneDB looking for new messages to send. When criteria are
+        // Forever, scan HyperlaneRocksDB looking for new messages to send. When criteria are
         // satisfied or the message is disqualified, push the message onto
         // self.tx_msg and then continue the scan at the next highest
         // nonce.

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

@@ -44,7 +44,7 @@ type OpQueue = Arc<Mutex<BinaryHeap<Reverse<Box<DynPendingOperation>>>>>;
 ///
 /// Finally, the SerialSubmitter ensures that message delivery is robust to
 /// destination chain reorgs prior to committing delivery status to
-/// HyperlaneDB.
+/// HyperlaneRocksDB.
 ///
 ///
 /// Objectives

rust/agents/relayer/src/relayer.rs

@@ -6,6 +6,7 @@ use std::{
 
 use async_trait::async_trait;
 use eyre::Result;
+use hyperlane_base::{MessageContractSync, WatermarkContractSync};
 use tokio::sync::mpsc::UnboundedSender;
 use tokio::{
     sync::{
@@ -17,11 +18,10 @@ use tokio::{
 use tracing::{info, info_span, instrument::Instrumented, Instrument};
 
 use hyperlane_base::{
-    db::{HyperlaneDB, DB},
-    run_all, BaseAgent, CachingInterchainGasPaymaster, CachingMailbox, ContractSyncMetrics,
-    CoreMetrics, HyperlaneAgentCore,
+    db::{HyperlaneRocksDB, DB},
+    run_all, BaseAgent, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
 };
-use hyperlane_core::{HyperlaneDomain, U256};
+use hyperlane_core::{HyperlaneDomain, InterchainGasPayment, U256};
 
 use crate::msg::pending_message::MessageSubmissionMetrics;
 use crate::{
@@ -48,15 +48,14 @@ pub struct Relayer {
     origin_chains: HashSet<HyperlaneDomain>,
     destination_chains: HashSet<HyperlaneDomain>,
     core: HyperlaneAgentCore,
+    message_syncs: HashMap<HyperlaneDomain, Arc<MessageContractSync>>,
+    interchain_gas_payment_syncs:
+        HashMap<HyperlaneDomain, Arc<WatermarkContractSync<InterchainGasPayment>>>,
     /// Context data for each (origin, destination) chain pair a message can be
     /// sent between
     msg_ctxs: HashMap<ContextKey, Arc<MessageContext>>,
-    /// Mailboxes for all chains (technically only need caching mailbox for
-    /// origin chains)
-    mailboxes: HashMap<HyperlaneDomain, CachingMailbox>,
-    /// Interchain gas paymaster for each origin chain
-    interchain_gas_paymasters: HashMap<HyperlaneDomain, CachingInterchainGasPaymaster>,
     prover_syncs: HashMap<HyperlaneDomain, Arc<RwLock<MerkleTreeBuilder>>>,
+    dbs: HashMap<HyperlaneDomain, HyperlaneRocksDB>,
     whitelist: Arc<MatchingList>,
     blacklist: Arc<MatchingList>,
     transaction_gas_limit: Option<U256>,
@@ -99,26 +98,40 @@ impl BaseAgent for Relayer {
     {
         let core = settings.build_hyperlane_core(metrics.clone());
         let db = DB::from_path(&settings.db)?;
-
-        // Use defined origin chains and remote chains
-        let domains = settings
+        let dbs = settings
             .origin_chains
             .iter()
-            .chain(&settings.destination_chains)
-            .collect::<HashSet<_>>();
+            .map(|origin| (origin.clone(), HyperlaneRocksDB::new(origin, db.clone())))
+            .collect::<HashMap<_, _>>();
 
         let mailboxes = settings
-            .build_all_mailboxes(domains.into_iter(), &metrics, db.clone())
+            .build_mailboxes(settings.destination_chains.iter(), &metrics)
             .await?;
-        let interchain_gas_paymasters = settings
-            .build_all_interchain_gas_paymasters(
+        let validator_announces = settings
+            .build_validator_announces(settings.origin_chains.iter(), &metrics)
+            .await?;
+
+        let contract_sync_metrics = Arc::new(ContractSyncMetrics::new(&metrics));
+
+        let message_syncs = settings
+            .build_message_indexers(
                 settings.origin_chains.iter(),
                 &metrics,
-                db.clone(),
+                &contract_sync_metrics,
+                dbs.iter()
+                    .map(|(d, db)| (d.clone(), Arc::new(db.clone()) as _))
+                    .collect(),
             )
             .await?;
-        let validator_announces = settings
-            .build_all_validator_announces(settings.origin_chains.iter(), &metrics)
+        let interchain_gas_payment_syncs = settings
+            .build_interchain_gas_payment_indexers(
+                settings.origin_chains.iter(),
+                &metrics,
+                &contract_sync_metrics,
+                dbs.iter()
+                    .map(|(d, db)| (d.clone(), Arc::new(db.clone()) as _))
+                    .collect(),
+            )
             .await?;
 
         let whitelist = Arc::new(settings.whitelist);
@@ -139,7 +152,7 @@ impl BaseAgent for Relayer {
             .origin_chains
             .iter()
             .map(|origin| {
-                let db = HyperlaneDB::new(origin, db.clone());
+                let db = dbs.get(origin).unwrap().clone();
                 (
                     origin.clone(),
                     Arc::new(RwLock::new(MerkleTreeBuilder::new(db))),
@@ -148,6 +161,7 @@ impl BaseAgent for Relayer {
             .collect::<HashMap<_, _>>();
 
         info!(gas_enforcement_policies=?settings.gas_payment_enforcement, "Gas enforcement configuration");
+
         // need one of these per origin chain due to the database scoping even though
         // the config itself is the same
         let gas_payment_enforcers: HashMap<_, _> = settings
@@ -158,7 +172,7 @@ impl BaseAgent for Relayer {
                     domain.clone(),
                     Arc::new(GasPaymentEnforcer::new(
                         settings.gas_payment_enforcement.clone(),
-                        HyperlaneDB::new(domain, db.clone()),
+                        dbs.get(domain).unwrap().clone(),
                     )),
                 )
             })
@@ -192,7 +206,7 @@ impl BaseAgent for Relayer {
                     },
                     Arc::new(MessageContext {
                         destination_mailbox: mailboxes[destination].clone(),
-                        origin_db: HyperlaneDB::new(origin, db.clone()),
+                        origin_db: dbs.get(origin).unwrap().clone(),
                         metadata_builder,
                         origin_gas_payment_enforcer: gas_payment_enforcers[origin].clone(),
                         transaction_gas_limit,
@@ -203,12 +217,13 @@ impl BaseAgent for Relayer {
         }
 
         Ok(Self {
+            dbs,
             origin_chains: settings.origin_chains,
             destination_chains: settings.destination_chains,
             msg_ctxs,
             core,
-            mailboxes,
-            interchain_gas_paymasters,
+            message_syncs,
+            interchain_gas_payment_syncs,
             prover_syncs,
             whitelist,
             blacklist,
@@ -232,10 +247,9 @@ impl BaseAgent for Relayer {
             tasks.push(self.run_destination_submitter(destination, receive_channel));
         }
 
-        let sync_metrics = ContractSyncMetrics::new(self.core.metrics.clone());
         for origin in &self.origin_chains {
-            tasks.push(self.run_origin_mailbox_sync(origin, sync_metrics.clone()));
-            tasks.push(self.run_interchain_gas_paymaster_sync(origin, sync_metrics.clone()));
+            tasks.push(self.run_message_sync(origin).await);
+            tasks.push(self.run_interchain_gas_payment_sync(origin).await);
         }
 
         // each message process attempts to send messages from a chain
@@ -248,28 +262,37 @@ impl BaseAgent for Relayer {
 }
 
 impl Relayer {
-    fn run_origin_mailbox_sync(
+    async fn run_message_sync(
         &self,
         origin: &HyperlaneDomain,
-        sync_metrics: ContractSyncMetrics,
-    ) -> Instrumented<JoinHandle<Result<()>>> {
-        let sync = self.mailboxes[origin].sync(
-            self.as_ref().settings.chains[origin.name()].index.clone(),
-            sync_metrics,
-        );
-        sync
+    ) -> Instrumented<JoinHandle<eyre::Result<()>>> {
+        let index_settings = self.as_ref().settings.chains[origin.name()].index.clone();
+        let contract_sync = self.message_syncs.get(origin).unwrap().clone();
+        let cursor = contract_sync
+            .forward_backward_message_sync_cursor(index_settings)
+            .await;
+        tokio::spawn(async move {
+            contract_sync
+                .clone()
+                .sync("dispatched_messages", cursor)
+                .await
+        })
+        .instrument(info_span!("ContractSync"))
     }
 
-    fn run_interchain_gas_paymaster_sync(
+    async fn run_interchain_gas_payment_sync(
         &self,
         origin: &HyperlaneDomain,
-        sync_metrics: ContractSyncMetrics,
-    ) -> Instrumented<JoinHandle<Result<()>>> {
-        let sync = self.interchain_gas_paymasters[origin].sync(
-            self.as_ref().settings.chains[origin.name()].index.clone(),
-            sync_metrics,
-        );
-        sync
+    ) -> Instrumented<JoinHandle<eyre::Result<()>>> {
+        let index_settings = self.as_ref().settings.chains[origin.name()].index.clone();
+        let contract_sync = self
+            .interchain_gas_payment_syncs
+            .get(origin)
+            .unwrap()
+            .clone();
+        let cursor = contract_sync.rate_limited_cursor(index_settings).await;
+        tokio::spawn(async move { contract_sync.clone().sync("gas_payments", cursor).await })
+            .instrument(info_span!("ContractSync"))
     }
 
     fn run_message_processor(
@@ -298,7 +321,7 @@ impl Relayer {
             })
             .collect();
         let message_processor = MessageProcessor::new(
-            self.mailboxes[origin].db().clone(),
+            self.dbs.get(origin).unwrap().clone(),
             self.whitelist.clone(),
             self.blacklist.clone(),
             metrics,