Process messages via Gelato ForwardRequests without charging gas yet (#967)

* WIP, messy, just wanna test

* Rename gelato_conf to gelato

* Enabled is a str, not a bool

* testing

* working at least

* closer

* log

* Things working pretty well

* Cargo fmt

* metrics

* cleanin

* cleaning

* clippy

* nit

* deploy tooling

* Fix disabled logic

* Instrument

* nits

* More sane gelato config

* nit

* Reset rc

* PR comment

* Add ForwardRequestOpArgs

* Add some chains to Gelato

* Use Http

* Fix test

rust/abacus-base/src/settings/chains.rs

@@ -36,7 +36,7 @@ impl Default for ChainConf {
 #[serde(rename_all = "camelCase")]
 pub struct GelatoConf {
     /// Whether to use the Gelato Relay service for transactions submitted to the chain.
-    pub enabled: bool,
+    pub enabled: String,
 }
 
 /// Addresses for outbox chain contracts
@@ -76,7 +76,7 @@ pub struct ChainSetup<T> {
     #[serde(flatten)]
     pub chain: ChainConf,
     /// Gelato configuration for this chain (Gelato unused if None)
-    pub gelato_conf: Option<GelatoConf>,
+    pub gelato: Option<GelatoConf>,
     /// Set this key to disable the inbox. Does nothing for outboxes.
     #[serde(default)]
     pub disabled: Option<String>,

rust/abacus-base/src/settings/mod.rs

@@ -260,7 +260,12 @@ impl Settings {
         metrics: &CoreMetrics,
     ) -> Result<HashMap<String, InboxContracts>, Report> {
         let mut result = HashMap::new();
-        for (k, v) in self.inboxes.iter().filter(|(_, v)| v.disabled.is_none()) {
+        for (k, v) in self.inboxes.iter().filter(|(_, v)| {
+            !v.disabled
+                .as_ref()
+                .and_then(|d| d.parse::<bool>().ok())
+                .unwrap_or(false)
+        }) {
             if k != &v.name {
                 bail!(
                     "Inbox key does not match inbox name:\n key: {}  name: {}",

rust/abacus-base/src/validator_manager.rs

@@ -1,5 +1,4 @@
 use async_trait::async_trait;
-use ethers::{prelude::AbiError, types::Bytes};
 use std::sync::Arc;
 
 use abacus_core::{
@@ -76,7 +75,7 @@ impl InboxValidatorManager for InboxValidatorManagerVariants {
         multisig_signed_checkpoint: &MultisigSignedCheckpoint,
         message: &AbacusMessage,
         proof: &Proof,
-    ) -> Result<Bytes, AbiError> {
+    ) -> Vec<u8> {
         match self {
             InboxValidatorManagerVariants::Ethereum(validator_manager) => {
                 validator_manager.process_calldata(multisig_signed_checkpoint, message, proof)

rust/abacus-core/src/traits/validator_manager.rs

@@ -1,7 +1,6 @@
 use std::fmt::Debug;
 
 use async_trait::async_trait;
-use ethers::{prelude::AbiError, types::Bytes};
 use eyre::Result;
 
 use crate::{
@@ -28,7 +27,7 @@ pub trait InboxValidatorManager: Send + Sync + Debug {
         multisig_signed_checkpoint: &MultisigSignedCheckpoint,
         message: &AbacusMessage,
         proof: &Proof,
-    ) -> Result<Bytes, AbiError>;
+    ) -> Vec<u8>;
 
     /// The on-chain address of the inbox validator manager contract.
     fn contract_address(&self) -> Address;

rust/agents/relayer/src/msg/gelato_submitter/fwd_req_op.rs

@@ -1,22 +1,244 @@
-use std::time::Duration;
+use std::{ops::Deref, sync::Arc, time::Duration};
 
+use abacus_base::InboxContracts;
+use abacus_core::{ChainCommunicationError, Inbox, InboxValidatorManager, MessageStatus};
+use ethers::{
+    signers::Signer,
+    types::{H160, U256},
+};
 use eyre::Result;
-use gelato::fwd_req_call::ForwardRequestArgs;
+use gelato::{
+    chains::Chain,
+    fwd_req_call::{
+        ForwardRequestArgs, ForwardRequestCall, ForwardRequestCallResult, PaymentType,
+        NATIVE_FEE_TOKEN_ADDRESS,
+    },
+    task_status_call::{TaskStatus, TaskStatusCall, TaskStatusCallArgs},
+};
+use tokio::{
+    sync::mpsc::UnboundedSender,
+    time::{sleep, timeout},
+};
+use tracing::instrument;
+
+use crate::msg::SubmitMessageArgs;
+
+/// The max fee to use for Gelato ForwardRequests.
+/// Gelato isn't charging fees on testnet. For now, use this hardcoded value
+/// of 1e18, or 1.0 ether.
+/// TODO: revisit before running on mainnet and when we consider interchain
+/// gas payments.
+const DEFAULT_MAX_FEE: u64 = 10u64.pow(18);
+
+/// The default gas limit to use for Gelato ForwardRequests, arbitrarily chose
+/// to be 5M.
+/// TODO: once Gelato fully deploys their new version, simply omit the gas
+/// limit so that Gelato does the estimation for us.
+const DEFAULT_GAS_LIMIT: u64 = 5000000;
 
-// TODO(webbhorn): Remove 'allow unused' once we impl run() and ref internal fields.
-#[allow(unused)]
 #[derive(Debug, Clone)]
-pub(crate) struct ForwardRequestOp<S> {
-    pub args: ForwardRequestArgs,
+pub struct ForwardRequestOpArgs<S> {
     pub opts: ForwardRequestOptions,
-    pub signer: S,
     pub http: reqwest::Client,
+
+    pub message: SubmitMessageArgs,
+    pub inbox_contracts: InboxContracts,
+    pub sponsor_signer: S,
+    pub sponsor_address: H160,
+    // Currently unused due to a bug in Gelato's testnet relayer that is currently being upgraded.
+    pub sponsor_chain: Chain,
+    pub destination_chain: Chain,
+
+    /// A channel to send the message over upon the message being successfully processed.
+    pub message_processed_sender: UnboundedSender<SubmitMessageArgs>,
+}
+
+#[derive(Debug, Clone)]
+pub struct ForwardRequestOp<S>(ForwardRequestOpArgs<S>);
+
+impl<S> Deref for ForwardRequestOp<S> {
+    type Target = ForwardRequestOpArgs<S>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
 }
 
-impl<S> ForwardRequestOp<S> {
-    #[allow(unused)]
-    pub async fn run(&self) -> Result<()> {
-        todo!()
+impl<S> ForwardRequestOp<S>
+where
+    S: Signer,
+    S::Error: 'static,
+{
+    pub fn new(args: ForwardRequestOpArgs<S>) -> Self {
+        Self(args)
+    }
+
+    #[instrument(skip(self), fields(msg_leaf_index=self.0.message.leaf_index))]
+    pub async fn run(&mut self) {
+        loop {
+            match self.tick().await {
+                Ok(MessageStatus::Processed) => {
+                    // If the message was processed, send it over the channel and
+                    // stop running.
+                    if let Err(err) = self.send_message_processed() {
+                        tracing::error!(
+                            err=?err,
+                            "Unable to send processed message, receiver is closed or dropped.",
+                        );
+                    }
+                    return;
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        err=?err,
+                        "Error occurred in fwd_req_op tick",
+                    );
+                }
+                _ => {}
+            }
+
+            self.0.message.num_retries += 1;
+            sleep(Duration::from_secs(5)).await;
+        }
+    }
+
+    async fn tick(&self) -> Result<MessageStatus> {
+        // Before doing anything, first check if the message has already been processed.
+        if let Ok(MessageStatus::Processed) = self.message_status().await {
+            return Ok(MessageStatus::Processed);
+        }
+
+        // Send the forward request.
+        let fwd_req_result = self.send_forward_request_call().await?;
+        tracing::info!(
+            msg=?self.0.message,
+            task_id=fwd_req_result.task_id,
+            "Sent forward request",
+        );
+
+        // Wait for a terminal state, timing out according to the retry_submit_interval.
+        match timeout(
+            self.0.opts.retry_submit_interval,
+            self.poll_for_terminal_state(fwd_req_result.task_id.clone()),
+        )
+        .await
+        {
+            Ok(result) => {
+                // Bubble up any error that may have occurred in `poll_for_terminal_state`.
+                result
+            }
+            // If a timeout occurred, don't bubble up an error, instead just log
+            // and set ourselves up for the next tick.
+            Err(err) => {
+                tracing::debug!(err=?err, "Forward request timed out, reattempting");
+                Ok(MessageStatus::None)
+            }
+        }
+    }
+
+    // Waits until the message has either been processed or the task id has been cancelled
+    // by Gelato.
+    async fn poll_for_terminal_state(&self, task_id: String) -> Result<MessageStatus> {
+        loop {
+            sleep(self.0.opts.poll_interval).await;
+
+            // Check if the message has been processed. Checking with the Inbox directly
+            // is the best source of truth, and is the only way in which a message can be
+            // marked as processed.
+            if let Ok(MessageStatus::Processed) = self.message_status().await {
+                return Ok(MessageStatus::Processed);
+            }
+
+            // Get the status of the ForwardRequest task from Gelato for debugging.
+            // If the task was cancelled for some reason by Gelato, stop waiting.
+
+            let status_call = TaskStatusCall {
+                http: Arc::new(self.0.http.clone()),
+                args: TaskStatusCallArgs {
+                    task_id: task_id.clone(),
+                },
+            };
+            let status_result = status_call.run().await?;
+
+            if let [tx_status] = &status_result.data[..] {
+                tracing::info!(
+                    task_id=task_id,
+                    tx_status=?tx_status,
+                    "Polled forward request status",
+                );
+
+                // The only terminal state status is if the task was cancelled, which happens after
+                // Gelato has known about the task for ~20 minutes and could not execute it.
+                if let TaskStatus::Cancelled = tx_status.task_state {
+                    return Ok(MessageStatus::None);
+                }
+            } else {
+                tracing::warn!(
+                    task_id=task_id,
+                    status_result_data=?status_result.data,
+                    "Unexpected forward request status data",
+                );
+            }
+        }
+    }
+
+    // Once gas payments are enforced, we will likely fetch the gas payment from
+    // the DB here. This is why forward request args are created and signed for each
+    // forward request call.
+    async fn send_forward_request_call(&self) -> Result<ForwardRequestCallResult> {
+        let args = self.create_forward_request_args();
+        let signature = self.0.sponsor_signer.sign_typed_data(&args).await?;
+
+        let fwd_req_call = ForwardRequestCall {
+            args,
+            http: self.0.http.clone(),
+            signature,
+        };
+
+        Ok(fwd_req_call.run().await?)
+    }
+
+    fn create_forward_request_args(&self) -> ForwardRequestArgs {
+        let calldata = self.0.inbox_contracts.validator_manager.process_calldata(
+            &self.0.message.checkpoint,
+            &self.0.message.committed_message.message,
+            &self.0.message.proof,
+        );
+        ForwardRequestArgs {
+            chain_id: self.0.destination_chain,
+            target: self
+                .inbox_contracts
+                .validator_manager
+                .contract_address()
+                .into(),
+            data: calldata.into(),
+            fee_token: NATIVE_FEE_TOKEN_ADDRESS,
+            payment_type: PaymentType::AsyncGasTank,
+            max_fee: DEFAULT_MAX_FEE.into(),
+            gas: DEFAULT_GAS_LIMIT.into(),
+            // At the moment, there's a bug with Gelato where environments that don't charge
+            // fees (i.e. testnet) require the sponsor chain ID to be the same as the chain
+            // in which the tx will be sent to.
+            // This will be fixed in an upcoming release they're doing.
+            sponsor_chain_id: self.0.destination_chain,
+            nonce: U256::zero(),
+            enforce_sponsor_nonce: false,
+            enforce_sponsor_nonce_ordering: false,
+            sponsor: self.0.sponsor_address,
+        }
+    }
+
+    async fn message_status(&self) -> Result<MessageStatus, ChainCommunicationError> {
+        self.inbox_contracts
+            .inbox
+            .message_status(self.message.committed_message.to_leaf())
+            .await
+    }
+
+    fn send_message_processed(
+        &self,
+    ) -> Result<(), tokio::sync::mpsc::error::SendError<SubmitMessageArgs>> {
+        self.message_processed_sender.send(self.message.clone())
     }
 }
 

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

@@ -1,59 +1,48 @@
-use std::collections::VecDeque;
-
 use abacus_base::{CoreMetrics, InboxContracts};
+use abacus_core::AbacusCommon;
 use abacus_core::{db::AbacusDB, Signers};
-use abacus_core::{AbacusCommon, InboxValidatorManager};
 use ethers::signers::Signer;
-use ethers::types::{Address, U256};
+use ethers::types::Address;
 use eyre::{bail, Result};
 use gelato::chains::Chain;
-use gelato::fwd_req_call::{ForwardRequestArgs, PaymentType, NATIVE_FEE_TOKEN_ADDRESS};
 use prometheus::{Histogram, IntCounter, IntGauge};
-use tokio::sync::mpsc;
-use tokio::time::{sleep, Duration};
+use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
+use tokio::time::{sleep, Duration, Instant};
 use tokio::{sync::mpsc::error::TryRecvError, task::JoinHandle};
 use tracing::{info_span, instrument::Instrumented, Instrument};
 
-use self::fwd_req_op::{ForwardRequestOp, ForwardRequestOptions};
+use crate::msg::gelato_submitter::fwd_req_op::{
+    ForwardRequestOp, ForwardRequestOpArgs, ForwardRequestOptions,
+};
 
 use super::SubmitMessageArgs;
 
 mod fwd_req_op;
 
-/// The max fee to use for Gelato ForwardRequests.
-/// Gelato isn't charging fees on testnet. For now, use this hardcoded value
-/// of 1e18, or 1.0 ether.
-/// TODO: revisit when testing on mainnet and actually considering interchain
-/// gas payments.
-const DEFAULT_MAX_FEE: u64 = 1000000000000000000;
-
-/// The default gas limit to use for Gelato ForwardRequests.
-/// TODO: instead estimate gas for messages.
-const DEFAULT_GAS_LIMIT: u64 = 3000000;
-
 #[derive(Debug)]
 pub(crate) struct GelatoSubmitter {
     /// Source of messages to submit.
-    pub message_receiver: mpsc::UnboundedReceiver<SubmitMessageArgs>,
+    message_receiver: mpsc::UnboundedReceiver<SubmitMessageArgs>,
     /// Inbox / InboxValidatorManager on the destination chain.
-    pub inbox_contracts: InboxContracts,
+    inbox_contracts: InboxContracts,
     /// The outbox chain in the format expected by the Gelato crate.
-    pub outbox_gelato_chain: Chain,
+    outbox_gelato_chain: Chain,
     /// The inbox chain in the format expected by the Gelato crate.
-    pub inbox_gelato_chain: Chain,
+    inbox_gelato_chain: Chain,
     /// The signer of the Gelato sponsor, used for EIP-712 meta-transaction signatures.
-    pub gelato_sponsor_signer: Signers,
+    gelato_sponsor_signer: Signers,
     /// The address of the Gelato sponsor.
-    pub gelato_sponsor_address: Address,
-    /// Messages we are aware of that we want to eventually submit, but haven't yet, for
-    /// whatever reason.
-    pub wait_queue: VecDeque<SubmitMessageArgs>,
+    gelato_sponsor_address: Address,
     /// Interface to agent rocks DB for e.g. writing delivery status upon completion.
-    pub _abacus_db: AbacusDB,
+    db: AbacusDB,
     /// Shared reqwest HTTP client to use for any ops to Gelato endpoints.
-    pub http_client: reqwest::Client,
+    http_client: reqwest::Client,
     /// Prometheus metrics.
-    pub _metrics: GelatoSubmitterMetrics,
+    metrics: GelatoSubmitterMetrics,
+    /// Channel used by ForwardRequestOps to send that their message has been successfully processed.
+    message_processed_sender: UnboundedSender<SubmitMessageArgs>,
+    /// Channel to receive from ForwardRequestOps that a message has been successfully processed.
+    message_processed_receiver: UnboundedReceiver<SubmitMessageArgs>,
 }
 
 impl GelatoSubmitter {
@@ -66,18 +55,21 @@ impl GelatoSubmitter {
         http_client: reqwest::Client,
         metrics: GelatoSubmitterMetrics,
     ) -> Self {
+        let (message_processed_sender, message_processed_receiver) =
+            mpsc::unbounded_channel::<SubmitMessageArgs>();
         Self {
             message_receiver,
             outbox_gelato_chain: abacus_domain_to_gelato_chain(outbox_domain).unwrap(),
             inbox_gelato_chain: abacus_domain_to_gelato_chain(inbox_contracts.inbox.local_domain())
                 .unwrap(),
             inbox_contracts,
-            _abacus_db: abacus_db,
+            db: abacus_db,
             gelato_sponsor_address: gelato_sponsor_signer.address(),
             gelato_sponsor_signer,
             http_client,
-            _metrics: metrics,
-            wait_queue: VecDeque::new(),
+            metrics,
+            message_processed_sender,
+            message_processed_receiver,
         }
     }
 
@@ -111,77 +103,72 @@ impl GelatoSubmitter {
             }
         }
 
-        // Insert received messages into the front of the wait queue, ensuring
-        // the asc ordering by message leaf index is preserved.
-        for msg in received_messages.into_iter().rev() {
-            self.wait_queue.push_front(msg);
+        // Spawn a ForwardRequestOp for each received message.
+        for msg in received_messages.into_iter() {
+            tracing::info!(msg=?msg, "Spawning forward request op for message");
+            let mut op = ForwardRequestOp::new(ForwardRequestOpArgs {
+                opts: ForwardRequestOptions::default(),
+                http: self.http_client.clone(),
+                message: msg,
+                inbox_contracts: self.inbox_contracts.clone(),
+                sponsor_signer: self.gelato_sponsor_signer.clone(),
+                sponsor_address: self.gelato_sponsor_address,
+                sponsor_chain: self.outbox_gelato_chain,
+                destination_chain: self.inbox_gelato_chain,
+                message_processed_sender: self.message_processed_sender.clone(),
+            });
+            self.metrics.active_forward_request_ops_gauge.add(1);
+
+            tokio::spawn(async move { op.run().await });
         }
 
-        // TODO: correctly process the wait queue.
-        // Messages should be popped from the wait queue. For messages
-        // with successful gas estimation, a ForwardRequestOp should
-        // be created. Messages whose gas estimation reverts should be
-        // pushed to the back of the queue.
-
-        // Pick the next message to try processing.
-        let msg = match self.wait_queue.pop_front() {
-            Some(m) => m,
-            None => return Ok(()),
-        };
-
-        let op = ForwardRequestOp {
-            args: self.create_forward_request_args(msg)?,
-            opts: ForwardRequestOptions::default(),
-            signer: self.gelato_sponsor_signer.clone(),
-            http: self.http_client.clone(),
-        };
-
-        tokio::spawn(async move {
-            op.run()
-                .await
-                .expect("failed unimplemented forward request submit op");
-        });
+        // Pull any messages that have been successfully processed by ForwardRequestOps
+        loop {
+            match self.message_processed_receiver.try_recv() {
+                Ok(msg) => {
+                    self.record_message_process_success(&msg)?;
+                }
+                Err(TryRecvError::Empty) => {
+                    break;
+                }
+                Err(_) => {
+                    bail!("Disconnected receive channel or fatal err");
+                }
+            }
+        }
 
         Ok(())
     }
 
-    fn create_forward_request_args(&self, msg: SubmitMessageArgs) -> Result<ForwardRequestArgs> {
-        let calldata = self.inbox_contracts.validator_manager.process_calldata(
-            &msg.checkpoint,
-            &msg.committed_message.message,
-            &msg.proof,
-        )?;
-        Ok(ForwardRequestArgs {
-            chain_id: self.inbox_gelato_chain,
-            target: self
-                .inbox_contracts
-                .validator_manager
-                .contract_address()
-                .into(),
-            data: calldata,
-            fee_token: NATIVE_FEE_TOKEN_ADDRESS,
-            payment_type: PaymentType::AsyncGasTank,
-            max_fee: DEFAULT_MAX_FEE.into(),
-            gas: DEFAULT_GAS_LIMIT.into(),
-            sponsor_chain_id: self.outbox_gelato_chain,
-            nonce: U256::zero(),
-            enforce_sponsor_nonce: false,
-            enforce_sponsor_nonce_ordering: false,
-            sponsor: self.gelato_sponsor_address,
-        })
+    /// Record in AbacusDB and various metrics that this process has observed the successful
+    /// processing of a message. An Ok(()) value returned by this function is the 'commit' point
+    /// in a message's lifetime for final processing -- after this function has been seen to
+    /// return 'Ok(())', then without a wiped AbacusDB, we will never re-attempt processing for
+    /// this message again, even after the relayer restarts.
+    fn record_message_process_success(&mut self, msg: &SubmitMessageArgs) -> Result<()> {
+        tracing::info!(msg=?msg, "Recording message as successfully processed");
+        self.db.mark_leaf_as_processed(msg.leaf_index)?;
+
+        self.metrics.active_forward_request_ops_gauge.sub(1);
+        self.metrics
+            .queue_duration_hist
+            .observe((Instant::now() - msg.enqueue_time).as_secs_f64());
+        self.metrics.highest_submitted_leaf_index =
+            std::cmp::max(self.metrics.highest_submitted_leaf_index, msg.leaf_index);
+        self.metrics
+            .processed_gauge
+            .set(self.metrics.highest_submitted_leaf_index as i64);
+        self.metrics.messages_processed_count.inc();
+        Ok(())
     }
 }
 
-// TODO(tkporter): Drop allow dead code directive once we handle
-// updating each of these metrics.
-#[allow(dead_code)]
 #[derive(Debug)]
 pub(crate) struct GelatoSubmitterMetrics {
-    wait_queue_length_gauge: IntGauge,
     queue_duration_hist: Histogram,
     processed_gauge: IntGauge,
     messages_processed_count: IntCounter,
-
+    active_forward_request_ops_gauge: IntGauge,
     /// Private state used to update actual metrics each tick.
     highest_submitted_leaf_index: u32,
 }
@@ -189,11 +176,6 @@ pub(crate) struct GelatoSubmitterMetrics {
 impl GelatoSubmitterMetrics {
     pub fn new(metrics: &CoreMetrics, outbox_chain: &str, inbox_chain: &str) -> Self {
         Self {
-            wait_queue_length_gauge: metrics.submitter_queue_length().with_label_values(&[
-                outbox_chain,
-                inbox_chain,
-                "wait_queue",
-            ]),
             queue_duration_hist: metrics
                 .submitter_queue_duration_histogram()
                 .with_label_values(&[outbox_chain, inbox_chain]),
@@ -205,6 +187,9 @@ impl GelatoSubmitterMetrics {
                 outbox_chain,
                 inbox_chain,
             ]),
+            active_forward_request_ops_gauge: metrics
+                .submitter_queue_length()
+                .with_label_values(&[outbox_chain, inbox_chain, "active_forward_request_ops"]),
             highest_submitted_leaf_index: 0,
         }
     }

rust/agents/relayer/src/relayer.rs

@@ -93,7 +93,7 @@ impl BaseAgent for Relayer {
             tasks.push(self.run_inbox(
                 inbox_contracts.clone(),
                 signed_checkpoint_receiver.clone(),
-                self.core.settings.inboxes[inbox_name].gelato_conf.as_ref(),
+                self.core.settings.inboxes[inbox_name].gelato.as_ref(),
                 signer,
             ));
         }
@@ -174,7 +174,7 @@ impl Relayer {
         &self,
         inbox_contracts: InboxContracts,
         signed_checkpoint_receiver: watch::Receiver<Option<MultisigSignedCheckpoint>>,
-        gelato_conf: Option<&GelatoConf>,
+        gelato: Option<&GelatoConf>,
         signer: Signers,
     ) -> Instrumented<JoinHandle<Result<()>>> {
         let outbox = self.outbox().outbox();
@@ -185,8 +185,8 @@ impl Relayer {
             inbox_contracts.inbox.chain_name(),
         );
         let (msg_send, msg_receive) = mpsc::unbounded_channel();
-        let submit_fut = match gelato_conf {
-            Some(cfg) if cfg.enabled => self
+        let submit_fut = match gelato {
+            Some(cfg) if cfg.enabled.parse::<bool>().unwrap() => self
                 .make_gelato_submitter_for_inbox(msg_receive, inbox_contracts.clone(), signer)
                 .spawn(),
             _ => {

rust/chains/abacus-ethereum/src/validator_manager.rs

@@ -5,7 +5,7 @@ use std::fmt::Display;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use ethers::abi::Token;
+use ethers::abi::AbiEncode;
 use ethers::prelude::*;
 use eyre::Result;
 
@@ -14,10 +14,14 @@ use abacus_core::{
     InboxValidatorManager, MultisigSignedCheckpoint, TxOutcome,
 };
 
-use crate::contracts::inbox_validator_manager::InboxValidatorManager as EthereumInboxValidatorManagerInternal;
+use crate::contracts::inbox_validator_manager::{
+    InboxValidatorManager as EthereumInboxValidatorManagerInternal, ProcessCall,
+};
 use crate::trait_builder::MakeableWithProvider;
 use crate::tx::report_tx;
 
+pub use crate::contracts::inbox_validator_manager::INBOXVALIDATORMANAGER_ABI;
+
 impl<M> Display for EthereumInboxValidatorManagerInternal<M>
 where
     M: Middleware,
@@ -125,36 +129,28 @@ where
         multisig_signed_checkpoint: &MultisigSignedCheckpoint,
         message: &AbacusMessage,
         proof: &Proof,
-    ) -> Result<Bytes, AbiError> {
-        self.contract.encode(
-            "process",
-            [
-                Token::Address(self.inbox_address),
-                Token::FixedBytes(
-                    multisig_signed_checkpoint
-                        .checkpoint
-                        .root
-                        .to_fixed_bytes()
-                        .into(),
-                ),
-                Token::Uint(multisig_signed_checkpoint.checkpoint.index.into()),
-                Token::Array(
-                    multisig_signed_checkpoint
-                        .signatures
-                        .iter()
-                        .map(|s| Token::Bytes(s.to_vec()))
-                        .collect(),
-                ),
-                Token::Bytes(message.to_vec()),
-                Token::FixedArray(
-                    proof.path[0..32]
-                        .iter()
-                        .map(|e| Token::FixedBytes(e.to_vec()))
-                        .collect(),
-                ),
-                Token::Uint(proof.index.into()),
-            ],
-        )
+    ) -> Vec<u8> {
+        let mut sol_proof: [[u8; 32]; 32] = Default::default();
+        sol_proof
+            .iter_mut()
+            .enumerate()
+            .for_each(|(i, elem)| *elem = proof.path[i].to_fixed_bytes());
+
+        let process_call = ProcessCall {
+            inbox: self.inbox_address,
+            root: multisig_signed_checkpoint.checkpoint.root.to_fixed_bytes(),
+            index: multisig_signed_checkpoint.checkpoint.index.into(),
+            signatures: multisig_signed_checkpoint
+                .signatures
+                .iter()
+                .map(|s| s.to_vec().into())
+                .collect(),
+            message: message.to_vec().into(),
+            proof: sol_proof,
+            leaf_index: proof.index.into(),
+        };
+
+        process_call.encode()
     }
 
     fn contract_address(&self) -> abacus_core::Address {

rust/gelato/src/chains.rs

@@ -90,10 +90,13 @@ impl Chain {
     // for Gelato-suppored chains, until a better / dynamic approach
     // becomes available.
     //
-    // See `getRelayForwarderAddrss()` in the SDK file
-    // https://github.com/gelatodigital/relay-sdk/blob/master/src/constants/index.ts.
+    // See `getRelayForwarderAddress()` in the SDK file
+    // https://github.com/gelatodigital/relay-sdk/blob/8a9b9b2d0ef92ea9a3d6d64a230d9467a4b4da6d/src/constants/index.ts#L87.
     pub fn relay_fwd_addr(&self) -> Result<Address, GelatoError> {
         match self {
+            Chain::Ethereum => Ok(Address::from_str(
+                "5ca448e53e77499222741DcB6B3c959Fa829dAf2",
+            )?),
             Chain::Rinkeby => Ok(Address::from_str(
                 "9B79b798563e538cc326D03696B3Be38b971D282",
             )?),
@@ -103,15 +106,26 @@ impl Chain {
             Chain::Kovan => Ok(Address::from_str(
                 "4F36f93F58d36DcbC1E60b9bdBE213482285C482",
             )?),
+
             Chain::Polygon => Ok(Address::from_str(
                 "c2336e796F77E4E57b6630b6dEdb01f5EE82383e",
             )?),
             Chain::PolygonMumbai => Ok(Address::from_str(
                 "3428E19A01E40333D5D51465A08476b8F61B86f3",
             )?),
+
             Chain::BinanceSmartChain => Ok(Address::from_str(
-                "247A1306b6122ba28862b19a95004899db91f1b5",
+                "eeea839E2435873adA11d5dD4CAE6032742C0445",
+            )?),
+
+            Chain::Alfajores => Ok(Address::from_str(
+                "c2336e796F77E4E57b6630b6dEdb01f5EE82383e",
+            )?),
+
+            Chain::Avalanche => Ok(Address::from_str(
+                "3456E168d2D7271847808463D6D383D079Bd5Eaa",
             )?),
+
             _ => Err(GelatoError::UnknownRelayForwardAddress(*self)),
         }
     }
@@ -142,11 +156,27 @@ mod tests {
 
     #[test]
     fn contracts() {
-        assert!(!Chain::Ethereum.relay_fwd_addr().is_ok());
+        assert!(Chain::Ethereum.relay_fwd_addr().is_ok());
         assert!(Chain::Rinkeby.relay_fwd_addr().is_ok());
         assert!(Chain::Goerli.relay_fwd_addr().is_ok());
         assert!(Chain::Kovan.relay_fwd_addr().is_ok());
+
         assert!(Chain::Polygon.relay_fwd_addr().is_ok());
         assert!(Chain::PolygonMumbai.relay_fwd_addr().is_ok());
+
+        assert!(Chain::BinanceSmartChain.relay_fwd_addr().is_ok());
+        assert!(!Chain::BinanceSmartChainTestnet.relay_fwd_addr().is_ok());
+
+        assert!(!Chain::Celo.relay_fwd_addr().is_ok());
+        assert!(Chain::Alfajores.relay_fwd_addr().is_ok());
+
+        assert!(Chain::Avalanche.relay_fwd_addr().is_ok());
+        assert!(!Chain::AvalancheFuji.relay_fwd_addr().is_ok());
+
+        assert!(!Chain::Optimism.relay_fwd_addr().is_ok());
+        assert!(!Chain::OptimismKovan.relay_fwd_addr().is_ok());
+
+        assert!(!Chain::Arbitrum.relay_fwd_addr().is_ok());
+        assert!(!Chain::ArbitrumRinkeby.relay_fwd_addr().is_ok());
     }
 }

rust/gelato/src/fwd_req_call.rs

@@ -3,10 +3,9 @@ use crate::err::GelatoError;
 use ethers::types::{Address, Bytes, Signature, U256};
 use serde::ser::SerializeStruct;
 use serde::{Deserialize, Serialize, Serializer};
-use tracing::info;
 use tracing::instrument;
 
-const GATEWAY_URL: &str = "https://gateway.api.gelato.digital";
+const GATEWAY_URL: &str = "https://relay.gelato.digital";
 
 pub const NATIVE_FEE_TOKEN_ADDRESS: ethers::types::Address = Address::repeat_byte(0xEE);
 
@@ -30,7 +29,7 @@ pub struct ForwardRequestArgs {
 pub struct ForwardRequestCall {
     pub http: reqwest::Client,
     pub args: ForwardRequestArgs,
-    pub sig: Signature,
+    pub signature: Signature,
 }
 
 #[derive(Debug, Clone, Eq, PartialEq)]
@@ -48,11 +47,10 @@ impl ForwardRequestCall {
         );
         let http_args = HTTPArgs {
             args: self.args.clone(),
-            sig: self.sig,
+            signature: self.signature,
         };
-        info!(?url, ?http_args);
         let res = self.http.post(url).json(&http_args).send().await?;
-        let result: HTTPResult = res.json().await.unwrap();
+        let result: HTTPResult = res.json().await?;
         Ok(ForwardRequestCallResult::from(result))
     }
 }
@@ -60,7 +58,7 @@ impl ForwardRequestCall {
 #[derive(Debug)]
 struct HTTPArgs {
     args: ForwardRequestArgs,
-    sig: Signature,
+    signature: Signature,
 }
 
 #[derive(Debug, Clone, Deserialize, Eq, PartialEq)]
@@ -76,7 +74,7 @@ struct HTTPResult {
 //
 // In total, we have to make the following logical changes from the default serde serialization:
 //     *  add a new top-level dict field 'typeId', with const literal value 'ForwardRequest'.
-//     *  hoist the two struct members (`args` and `sig`) up to the top-level dict (equiv. to
+//     *  hoist the two struct members (`args` and `signature`) up to the top-level dict (equiv. to
 //        `#[serde(flatten)]`).
 //     *  make sure the integers for the fields `gas` and `maxfee` are enclosed within quotes,
 //        since Gelato-server-side, they will be interpreted as ~bignums.
@@ -107,7 +105,7 @@ impl Serialize for HTTPArgs {
             "enforceSponsorNonceOrdering",
             &self.args.enforce_sponsor_nonce_ordering,
         )?;
-        state.serialize_field("sponsorSignature", &format!("0x{}", self.sig))?;
+        state.serialize_field("sponsorSignature", &format!("0x{}", self.signature))?;
         state.end()
     }
 }
@@ -142,8 +140,8 @@ mod tests {
         let wallet = test_data::sdk_demo_data::WALLET_KEY
             .parse::<LocalWallet>()
             .unwrap();
-        let sig = wallet.sign_typed_data(&args).await.unwrap();
-        let http_args = HTTPArgs { args: args, sig };
+        let signature = wallet.sign_typed_data(&args).await.unwrap();
+        let http_args = HTTPArgs { args, signature };
         assert_eq!(
             serde_json::to_string(&http_args).unwrap(),
             test_data::sdk_demo_data::EXPECTED_JSON_REQUEST_CONTENT

rust/gelato/src/task_status_call.rs

@@ -1,7 +1,7 @@
 use crate::err::GelatoError;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tracing::{info, instrument};
+use tracing::instrument;
 
 const RELAY_URL: &str = "https://relay.gelato.digital";
 
@@ -26,9 +26,7 @@ impl TaskStatusCall {
     #[instrument]
     pub async fn run(&self) -> Result<TaskStatusCallResult, GelatoError> {
         let url = format!("{}/tasks/GelatoMetaBox/{}", RELAY_URL, self.args.task_id);
-        info!(?url);
         let res = self.http.get(url).send().await?;
-        info!(?res);
         let result: TaskStatusCallResult = res.json().await?;
         Ok(result)
     }
@@ -83,12 +81,16 @@ pub enum Check {
 }
 
 #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize)]
-#[serde(rename_all = "camelCase")]
 pub struct CheckInfo {
+    // See `created_at` to understand why we rename this field
+    // rather than using `#[serde(rename_all = "camelCase")].
+    #[serde(rename = "taskState")]
     pub task_state: TaskStatus,
     pub message: String,
-    pub payload: Payload,
+    pub payload: Option<Payload>,
     pub chain: Option<String>,
+    // Sadly, this is not serialized in camelCase by Gelato's API, and is
+    // named `created_at`.
     pub created_at: Option<String>,
 }
 

rust/helm/abacus-agent/templates/_helpers.tpl

@@ -74,23 +74,26 @@ The name of the ClusterSecretStore
 Recursively converts a config object into environment variables than can
 be parsed by rust. For example, a config of { foo: { bar: { baz: 420 }, boo: 421 } } will
 be: ABC_FOO_BAR_BAZ=420 and ABC_FOO_BOO=421
-Env vars can be formatted in FOO=BAR format if .dot_env_format is true, otherwise
-they will be formatted as YAML-friendly environment variables
+Env vars can be formatted in FOO="BAR" format if .format is "dot_env",
+FOO: "BAR" format if .format is "config_map", or otherwise
+they will be formatted as spec YAML-friendly environment variables
 */}}
 {{- define "abacus-agent.config-env-vars" -}}
 {{- range $key, $value := .config }}
 {{- $key_name := printf "%s%s" (default "" $.key_name_prefix) $key }}
 {{- if typeIs "map[string]interface {}" $value }}
-{{- include "abacus-agent.config-env-vars" (dict "config" $value "agent_name" $.agent_name "dot_env_format" $.dot_env_format "key_name_prefix" (printf "%s_" $key_name)) }}
+{{- include "abacus-agent.config-env-vars" (dict "config" $value "agent_name" $.agent_name "format" $.format "key_name_prefix" (printf "%s_" $key_name)) }}
 {{- else }}
-{{- include "abacus-agent.config-env-var" (dict "agent_name" $.agent_name "key" $key_name "value" $value "dot_env_format" $.dot_env_format ) }}
+{{- include "abacus-agent.config-env-var" (dict "agent_name" $.agent_name "key" $key_name "value" $value "format" $.format ) }}
 {{- end }}
 {{- end }}
 {{- end }}
 
 {{- define "abacus-agent.config-env-var" }}
-{{- if .dot_env_format }}
+{{- if (eq .format "dot_env") }}
 ABC_{{ .agent_name | upper }}_{{ .key | upper }}={{ .value | quote }}
+{{- else if (eq .format "config_map") }}
+ABC_{{ .agent_name | upper }}_{{ .key | upper }}: {{ .value | quote }}
 {{- else }}
 - name: ABC_{{ .agent_name | upper }}_{{ .key | upper }}
   value: {{ .value | quote }}

rust/helm/abacus-agent/templates/configmap.yaml

@@ -15,15 +15,7 @@ data:
   ABC_BASE_OUTBOX_CONNECTION_TYPE: {{ .Values.abacus.outboxChain.connectionType }}
   {{- end }}
   {{- range .Values.abacus.inboxChains }}
-  {{- if .address }}
-  ABC_BASE_INBOXES_{{ .name | upper }}_ADDRESS: {{ .address }}
-  {{- end }}
-  {{- if .disabled }}
-  ABC_BASE_INBOXES_{{ .name | upper }}_DISABLED: "true"
-  {{- end }}
-  {{- if .connectionType }}
-  ABC_BASE_INBOXES_{{ .name | upper }}_CONNECTION_TYPE: {{ .connectionType }}
-  {{- end }}
+{{- include "abacus-agent.config-env-vars" (dict "config" . "agent_name" "base" "key_name_prefix" (printf "INBOXES_%s_" (.name | upper)) "format" "config_map") | indent 2 }}
   {{- end }}
   {{- if .Values.abacus.tracing.uri }}
   ABC_BASE_TRACING_JAEGER_COLLECTOR_URI: {{ .Values.abacus.tracing.uri }}

rust/helm/abacus-agent/templates/validator-configmap.yaml

@@ -9,7 +9,7 @@ data:
 {{ $index := 0 }}
 {{- range .Values.abacus.validator.configs }}
   validator-{{ $index }}.env: |
-{{- include "abacus-agent.config-env-vars" (dict "config" . "agent_name" "validator" "dot_env_format" true) | indent 4 }}
+{{- include "abacus-agent.config-env-vars" (dict "config" . "agent_name" "validator" "format" "dot_env") | indent 4 }}
 {{ $index = add1 $index }}
 {{- end }}
 {{- end }}

typescript/infra/config/environments/testnet2/agent.ts

@@ -112,8 +112,12 @@ export const releaseCandidate: AgentConfig<TestnetChains> = {
   },
   environmentChainNames: chainNames,
   contextChainNames: chainNames,
+  gelato: {
+    enabledChains: ['alfajores', 'mumbai', 'kovan'],
+    useForDisabledOriginChains: true,
+  },
   validatorSets: validators,
-  connectionType: ConnectionType.HttpQuorum,
+  connectionType: ConnectionType.Http,
   relayer: {
     default: {
       signedCheckpointPollingInterval: 5,

typescript/infra/src/agents/index.ts

@@ -21,6 +21,11 @@ async function helmValuesForChain<Chain extends ChainName>(
 ) {
   const chainAgentConfig = new ChainAgentConfig(agentConfig, chainName);
 
+  const gelatoSupportedOnOutboxChain = agentConfig.gelato
+    ?.useForDisabledOriginChains
+    ? true
+    : agentConfig.gelato?.enabledChains.includes(chainName) ?? false;
+
   return {
     image: {
       repository: agentConfig.docker.repo,
@@ -41,7 +46,15 @@ async function helmValuesForChain<Chain extends ChainName>(
           return {
             name: remoteChainName,
             disabled: !agentConfig.contextChainNames.includes(remoteChainName),
-            connectionType: agentConfig.connectionType,
+            gelato: {
+              enabled:
+                gelatoSupportedOnOutboxChain &&
+                (agentConfig.gelato?.enabledChains?.includes(remoteChainName) ??
+                  false),
+            },
+            connection: {
+              type: agentConfig.connectionType,
+            },
           };
         }),
       validator: {

typescript/infra/src/config/agent.ts

@@ -178,6 +178,19 @@ export interface DockerConfig {
   tag: string;
 }
 
+export interface GelatoConfig<Chain extends ChainName> {
+  // List of chains in which using Gelato is enabled for
+  enabledChains: Chain[];
+  // If true, Gelato will still be used for messages whose
+  // origin chain is *not* supported by Gelato. If false,
+  // Gelato will not be used for any messages from a disabled
+  // origin chain, even if the destination chain is enabled.
+  // Because Gelato doesn't charge on testnets, this is likely
+  // to be true for testnet environments where the chain in which gas
+  // is paid on (the origin) doesn't need to be supported by Gelato.
+  useForDisabledOriginChains: boolean;
+}
+
 export interface AgentConfig<Chain extends ChainName> {
   environment: string;
   namespace: string;
@@ -193,6 +206,7 @@ export interface AgentConfig<Chain extends ChainName> {
   // Names of chains this context cares about
   contextChainNames: Chain[];
   validatorSets: ChainValidatorSets<Chain>;
+  gelato?: GelatoConfig<Chain>;
   validator?: ChainValidatorConfigs<Chain>;
   relayer?: ChainRelayerConfigs<Chain>;
   // Roles to manage keys for