V3 agents rebase

.github/workflows/e2e.yml

@@ -4,7 +4,7 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
+    branches: '*'
   workflow_dispatch:
 
 concurrency:

.gitignore

@@ -28,3 +28,4 @@ yarn-error.log
 **/*.ignore
 .vscode
 
+tsconfig.editor.json

rust/Cargo.toml

@@ -52,6 +52,7 @@ Inflector = "0.11.4"
 anyhow = "1.0"
 async-trait = "0.1"
 auto_impl = "1.0"
+backoff = { version = "0.4.0", features = ["tokio"] }
 backtrace = "0.3"
 base64 = "0.21.2"
 bincode = "1.3"

rust/agents/relayer/Cargo.toml

@@ -11,7 +11,9 @@ version.workspace = true
 
 [dependencies]
 async-trait.workspace = true
+backoff.workspace = true
 config.workspace = true
+convert_case.workspace = true
 derive-new.workspace = true
 derive_more.workspace = true
 enum_dispatch.workspace = true

rust/agents/relayer/src/main.rs

@@ -13,8 +13,9 @@ use hyperlane_base::agent_main;
 
 use crate::relayer::Relayer;
 
-mod merkle_tree_builder;
+mod merkle_tree;
 mod msg;
+mod processor;
 mod prover;
 mod relayer;
 mod settings;

rust/agents/relayer/src/merkle_tree_builder.rs → rust/agents/relayer/src/merkle_tree/builder.rs

@@ -1,6 +1,6 @@
 use std::fmt::Display;
 
-use eyre::Result;
+use eyre::{Context, Result};
 use tracing::{debug, error, instrument};
 
 use hyperlane_base::db::{DbError, HyperlaneRocksDB};
@@ -50,12 +50,6 @@ pub enum MerkleTreeBuilderError {
         /// Root of the incremental merkle tree
         incremental_root: H256,
     },
-    /// Nonce was not found in DB, despite batch providing messages after
-    #[error("Nonce was not found {nonce:?}")]
-    UnavailableNonce {
-        /// Root of prover's local merkle tree
-        nonce: u32,
-    },
     /// MerkleTreeBuilder attempts Prover operation and receives ProverError
     #[error(transparent)]
     ProverError(#[from] ProverError),
@@ -65,6 +59,9 @@ pub enum MerkleTreeBuilderError {
     /// DB Error
     #[error("{0}")]
     DbError(#[from] DbError),
+    /// Some other error occured.
+    #[error("Failed to build the merkle tree: {0}")]
+    Other(String),
 }
 
 impl MerkleTreeBuilder {
@@ -81,54 +78,38 @@ impl MerkleTreeBuilder {
     #[instrument(err, skip(self), level="debug", fields(prover_latest_index=self.count()-1))]
     pub fn get_proof(
         &self,
-        leaf_index: u32,
+        message_nonce: u32,
         root_index: u32,
-    ) -> Result<Proof, MerkleTreeBuilderError> {
+    ) -> Result<Option<Proof>, MerkleTreeBuilderError> {
+        let Some(leaf_index) = self
+            .db
+            .retrieve_message_id_by_nonce(&message_nonce)?
+            .and_then(|message_id| self.db.retrieve_merkle_leaf_index_by_message_id(&message_id).ok().flatten())
+        else {
+            return Ok(None);
+        };
         self.prover
             .prove_against_previous(leaf_index as usize, root_index as usize)
+            .map(Option::from)
             .map_err(Into::into)
     }
 
-    fn ingest_nonce(&mut self, nonce: u32) -> Result<(), MerkleTreeBuilderError> {
-        match self.db.retrieve_message_id_by_nonce(&nonce) {
-            Ok(Some(leaf)) => {
-                debug!(nonce, "Ingesting leaf");
-                self.prover.ingest(leaf).expect("!tree full");
-                self.incremental.ingest(leaf);
-                assert_eq!(self.prover.root(), self.incremental.root());
-                Ok(())
-            }
-            Ok(None) => {
-                error!("We should not arrive here");
-                Err(MerkleTreeBuilderError::UnavailableNonce { nonce })
-            }
-            Err(e) => Err(e.into()),
-        }
-    }
-
     pub fn count(&self) -> u32 {
         self.prover.count() as u32
     }
 
-    #[instrument(err, skip(self), level = "debug")]
-    pub async fn update_to_index(&mut self, index: u32) -> Result<(), MerkleTreeBuilderError> {
-        if index >= self.count() {
-            let starting_index = self.prover.count() as u32;
-            for i in starting_index..=index {
-                self.db.wait_for_message_nonce(i).await?;
-                self.ingest_nonce(i)?;
-            }
-
-            let prover_root = self.prover.root();
-            let incremental_root = self.incremental.root();
-            if prover_root != incremental_root {
-                return Err(MerkleTreeBuilderError::MismatchedRoots {
-                    prover_root,
-                    incremental_root,
-                });
-            }
+    pub async fn ingest_message_id(&mut self, message_id: H256) -> Result<()> {
+        const CTX: &str = "When ingesting message id";
+        debug!(?message_id, "Ingesting leaf");
+        self.prover.ingest(message_id).expect("tree full");
+        self.incremental.ingest(message_id);
+        match self.prover.root().eq(&self.incremental.root()) {
+            true => Ok(()),
+            false => Err(MerkleTreeBuilderError::MismatchedRoots {
+                prover_root: self.prover.root(),
+                incremental_root: self.incremental.root(),
+            }),
         }
-
-        Ok(())
+        .context(CTX)
     }
 }

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

@@ -0,0 +1,2 @@
+pub(crate) mod builder;
+pub(crate) mod processor;

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

@@ -0,0 +1,101 @@
+use std::{
+    fmt::{Debug, Formatter},
+    sync::Arc,
+    time::Duration,
+};
+
+use async_trait::async_trait;
+use derive_new::new;
+use eyre::Result;
+use hyperlane_base::db::HyperlaneRocksDB;
+use hyperlane_core::{HyperlaneDomain, MerkleTreeInsertion};
+use prometheus::IntGauge;
+use tokio::sync::RwLock;
+use tracing::debug;
+
+use crate::processor::ProcessorExt;
+
+use super::builder::MerkleTreeBuilder;
+
+/// Finds unprocessed merkle tree insertions and adds them to the prover sync
+#[derive(new)]
+pub struct MerkleTreeProcessor {
+    db: HyperlaneRocksDB,
+    metrics: MerkleTreeProcessorMetrics,
+    prover_sync: Arc<RwLock<MerkleTreeBuilder>>,
+    #[new(default)]
+    leaf_index: u32,
+}
+
+impl Debug for MerkleTreeProcessor {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MerkleTreeProcessor {{ leaf_index: {:?} }}",
+            self.leaf_index
+        )
+    }
+}
+
+#[async_trait]
+impl ProcessorExt for MerkleTreeProcessor {
+    /// The domain this processor is getting merkle tree hook insertions from.
+    fn domain(&self) -> &HyperlaneDomain {
+        self.db.domain()
+    }
+
+    /// One round of processing, extracted from infinite work loop for
+    /// testing purposes.
+    async fn tick(&mut self) -> Result<()> {
+        if let Some(insertion) = self.next_unprocessed_leaf()? {
+            // Feed the message to the prover sync
+            self.prover_sync
+                .write()
+                .await
+                .ingest_message_id(insertion.message_id())
+                .await?;
+
+            // Increase the leaf index to move on to the next leaf
+            self.leaf_index += 1;
+        } else {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+        Ok(())
+    }
+}
+
+impl MerkleTreeProcessor {
+    fn next_unprocessed_leaf(&mut self) -> Result<Option<MerkleTreeInsertion>> {
+        let leaf = if let Some(insertion) = self
+            .db
+            .retrieve_merkle_tree_insertion_by_leaf_index(&self.leaf_index)?
+        {
+            // Update the metrics
+            self.metrics
+                .max_leaf_index_gauge
+                .set(insertion.index() as i64);
+            Some(insertion)
+        } else {
+            debug!(leaf_index=?self.leaf_index, "No message found in DB for leaf index");
+            None
+        };
+        Ok(leaf)
+    }
+}
+
+#[derive(Debug)]
+pub struct MerkleTreeProcessorMetrics {
+    max_leaf_index_gauge: IntGauge,
+}
+
+impl MerkleTreeProcessorMetrics {
+    pub fn new() -> Self {
+        Self {
+            max_leaf_index_gauge: IntGauge::new(
+                "max_leaf_index_gauge",
+                "The max merkle tree leaf index",
+            )
+            .unwrap(),
+        }
+    }
+}

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

@@ -2,20 +2,20 @@ use std::fmt::Debug;
 
 use async_trait::async_trait;
 use eyre::Result;
-use tracing::{debug, error, trace};
-
 use hyperlane_base::db::HyperlaneRocksDB;
 use hyperlane_core::{
     GasPaymentKey, HyperlaneMessage, InterchainGasExpenditure, InterchainGasPayment,
     TxCostEstimate, TxOutcome, U256,
 };
-
-use crate::msg::gas_payment::policies::GasPaymentPolicyOnChainFeeQuoting;
-use crate::settings::{
-    matching_list::MatchingList, GasPaymentEnforcementConf, GasPaymentEnforcementPolicy,
-};
+use tracing::{debug, error, trace};
 
 use self::policies::{GasPaymentPolicyMinimum, GasPaymentPolicyNone};
+use crate::{
+    msg::gas_payment::policies::GasPaymentPolicyOnChainFeeQuoting,
+    settings::{
+        matching_list::MatchingList, GasPaymentEnforcementConf, GasPaymentEnforcementPolicy,
+    },
+};
 
 mod policies;
 
@@ -148,12 +148,11 @@ mod test {
         H256, U256,
     };
 
+    use super::GasPaymentEnforcer;
     use crate::settings::{
         matching_list::MatchingList, GasPaymentEnforcementConf, GasPaymentEnforcementPolicy,
     };
 
-    use super::GasPaymentEnforcer;
-
     #[tokio::test]
     async fn test_empty_whitelist() {
         test_utils::run_test_db(|db| async move {
@@ -195,7 +194,7 @@ mod test {
         test_utils::run_test_db(|db| async move {
             let hyperlane_db =
                 HyperlaneRocksDB::new(&HyperlaneDomain::new_test_domain("test_no_match"), db);
-            let matching_list = serde_json::from_str(r#"[{"originDomain": 234}]"#).unwrap();
+            let matching_list = serde_json::from_str(r#"[{"origindomain": 234}]"#).unwrap();
             let enforcer = GasPaymentEnforcer::new(
                 // Require a payment
                 vec![GasPaymentEnforcementConf {
@@ -339,7 +338,7 @@ mod test {
             let recipient_address = "0xbb000000000000000000000000000000000000bb";
 
             let matching_list = serde_json::from_str(
-                &format!(r#"[{{"senderAddress": "{sender_address}", "recipientAddress": "{recipient_address}"}}]"#)
+                &format!(r#"[{{"senderaddress": "{sender_address}", "recipientaddress": "{recipient_address}"}}]"#)
             ).unwrap();
 
             let enforcer = GasPaymentEnforcer::new(