Add Validator scaffold (#213)

rust/Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "chains/abacus-ethereum",
     "agents/kathy",
     "agents/checkpointer",
+    "agents/validator",
     "agents/updater",
     "agents/relayer",
     "agents/watcher",

rust/abacus-base/Cargo.toml

@@ -15,7 +15,7 @@ ethers = { git = "https://github.com/gakonst/ethers-rs", branch = "master" }
 thiserror = { version = "1.0.22", default-features = false }
 async-trait = { version = "0.1.42", default-features = false }
 futures-util = "0.3.12"
-color-eyre = "0.5.0"
+color-eyre = "0.6.1"
 tracing = "0.1.22"
 tracing-futures = "0.2.4"
 tracing-subscriber = "0.2.15"

rust/abacus-base/src/inbox.rs

@@ -2,7 +2,7 @@ use abacus_core::{
     accumulator::merkle::Proof, db::AbacusDB, AbacusMessage, ChainCommunicationError,
     MessageStatus, TxOutcome,
 };
-use abacus_core::{AbacusCommon, Inbox};
+use abacus_core::{AbacusCommon, Checkpoint, Inbox, SignedCheckpoint};
 use abacus_test::mocks::inbox::MockInboxContract;
 use async_trait::async_trait;
 use color_eyre::eyre::Result;
@@ -91,6 +91,13 @@ impl Inbox for CachingInbox {
     async fn message_status(&self, leaf: H256) -> Result<MessageStatus, ChainCommunicationError> {
         self.inbox.message_status(leaf).await
     }
+
+    async fn submit_checkpoint(
+        &self,
+        signed_checkpoint: &SignedCheckpoint,
+    ) -> Result<TxOutcome, ChainCommunicationError> {
+        self.inbox.submit_checkpoint(signed_checkpoint).await
+    }
 }
 
 #[async_trait]
@@ -114,6 +121,13 @@ impl AbacusCommon for CachingInbox {
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {
         self.inbox.checkpointed_root().await
     }
+
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        self.inbox.latest_checkpoint(maybe_lag).await
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -230,6 +244,19 @@ impl Inbox for InboxVariants {
             InboxVariants::Other(inbox) => inbox.prove_and_process(message, proof).await,
         }
     }
+
+    async fn submit_checkpoint(
+        &self,
+        signed_checkpoint: &SignedCheckpoint,
+    ) -> Result<TxOutcome, ChainCommunicationError> {
+        match self {
+            InboxVariants::Ethereum(inbox) => inbox.submit_checkpoint(signed_checkpoint).await,
+            InboxVariants::Mock(mock_inbox) => {
+                mock_inbox.submit_checkpoint(signed_checkpoint).await
+            }
+            InboxVariants::Other(inbox) => inbox.submit_checkpoint(signed_checkpoint).await,
+        }
+    }
 }
 
 #[async_trait]
@@ -273,4 +300,15 @@ impl AbacusCommon for InboxVariants {
             InboxVariants::Other(inbox) => inbox.checkpointed_root().await,
         }
     }
+
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        match self {
+            InboxVariants::Ethereum(inbox) => inbox.latest_checkpoint(maybe_lag).await,
+            InboxVariants::Mock(mock_inbox) => mock_inbox.latest_checkpoint(maybe_lag).await,
+            InboxVariants::Other(inbox) => inbox.latest_checkpoint(maybe_lag).await,
+        }
+    }
 }

rust/abacus-base/src/lib.rs

@@ -49,3 +49,9 @@ pub use contract_sync::*;
 
 mod indexer;
 pub use indexer::*;
+
+mod traits;
+pub use traits::*;
+
+mod types;
+pub use types::*;

rust/abacus-base/src/outbox.rs

@@ -1,7 +1,7 @@
 use abacus_core::db::AbacusDB;
 use abacus_core::{
-    AbacusCommon, ChainCommunicationError, Message, Outbox, OutboxEvents, RawCommittedMessage,
-    State, TxOutcome,
+    AbacusCommon, ChainCommunicationError, Checkpoint, Message, Outbox, OutboxEvents,
+    RawCommittedMessage, State, TxOutcome,
 };
 
 use abacus_ethereum::EthereumOutbox;
@@ -164,6 +164,13 @@ impl AbacusCommon for CachingOutbox {
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {
         self.outbox.checkpointed_root().await
     }
+
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        self.outbox.latest_checkpoint(maybe_lag).await
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -305,4 +312,15 @@ impl AbacusCommon for OutboxVariants {
             OutboxVariants::Other(outbox) => outbox.checkpointed_root().await,
         }
     }
+
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        match self {
+            OutboxVariants::Ethereum(outbox) => outbox.latest_checkpoint(maybe_lag).await,
+            OutboxVariants::Mock(mock_outbox) => mock_outbox.latest_checkpoint(maybe_lag).await,
+            OutboxVariants::Other(outbox) => outbox.latest_checkpoint(maybe_lag).await,
+        }
+    }
 }

rust/abacus-base/src/traits/checkpoint_syncer.rs

@@ -0,0 +1,14 @@
+use abacus_core::SignedCheckpoint;
+use async_trait::async_trait;
+use color_eyre::Result;
+
+/// A generic trait to read/write Checkpoints offchain
+#[async_trait]
+pub trait CheckpointSyncer {
+    /// Read the highest index of this Syncer
+    async fn latest_index(&self) -> Result<Option<u32>>;
+    /// Attempt to fetch the signed checkpoint at this index
+    async fn fetch_checkpoint(&self, index: u32) -> Result<Option<SignedCheckpoint>>;
+    /// Write the signed checkpoint to this syncer
+    async fn write_checkpoint(&self, signed_checkpoint: SignedCheckpoint) -> Result<()>;
+}

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

@@ -0,0 +1,3 @@
+mod checkpoint_syncer;
+
+pub use checkpoint_syncer::*;

rust/abacus-base/src/types/local_storage.rs

@@ -0,0 +1,76 @@
+use abacus_core::SignedCheckpoint;
+
+use async_trait::async_trait;
+use color_eyre::Result;
+
+use crate::traits::CheckpointSyncer;
+/// Type for reading/write to LocalStorage
+pub struct LocalStorage {
+    /// base path
+    pub path: String,
+}
+
+impl LocalStorage {
+    fn checkpoint_file_path(&self, index: u32) -> String {
+        let mut path = self.path.clone();
+        path.push_str(&format!("/{}.json", index));
+        path
+    }
+
+    fn latest_index_file_path(&self) -> String {
+        let mut path = self.path.clone();
+        path.push_str("/index.json");
+        path
+    }
+
+    async fn write_index(&self, index: u32) -> Result<()> {
+        tokio::fs::write(self.latest_index_file_path(), index.to_string()).await?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl CheckpointSyncer for LocalStorage {
+    async fn latest_index(&self) -> Result<Option<u32>> {
+        match tokio::fs::read(self.latest_index_file_path())
+            .await
+            .and_then(|data| {
+                String::from_utf8(data)
+                    .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))
+            }) {
+            Ok(data) => {
+                let index = data.parse()?;
+                Ok(Some(index))
+            }
+            _ => Ok(None),
+        }
+    }
+    async fn fetch_checkpoint(&self, index: u32) -> Result<Option<SignedCheckpoint>> {
+        match tokio::fs::read(self.checkpoint_file_path(index)).await {
+            Ok(data) => {
+                let checkpoint = serde_json::from_slice(&data)?;
+                Ok(Some(checkpoint))
+            }
+            _ => Ok(None),
+        }
+    }
+    async fn write_checkpoint(&self, signed_checkpoint: SignedCheckpoint) -> Result<()> {
+        let serialized_checkpoint = serde_json::to_string_pretty(&signed_checkpoint)?;
+        tokio::fs::write(
+            self.checkpoint_file_path(signed_checkpoint.checkpoint.index),
+            &serialized_checkpoint,
+        )
+        .await?;
+
+        match self.latest_index().await? {
+            Some(current_latest_index) => {
+                if current_latest_index < signed_checkpoint.checkpoint.index {
+                    self.write_index(signed_checkpoint.checkpoint.index).await?
+                }
+            }
+            None => self.write_index(signed_checkpoint.checkpoint.index).await?,
+        }
+
+        Ok(())
+    }
+}

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

@@ -0,0 +1,3 @@
+mod local_storage;
+
+pub use local_storage::*;

rust/abacus-core/src/lib.rs

@@ -53,11 +53,11 @@ pub use identifiers::AbacusIdentifier;
 
 use async_trait::async_trait;
 use ethers::{
-    core::types::{Address as EthAddress, Signature, SignatureError, H256},
-    prelude::{
+    core::types::{
         transaction::{eip2718::TypedTransaction, eip712::Eip712},
-        AwsSigner,
+        Address as EthAddress, Signature, SignatureError, H256,
     },
+    prelude::AwsSigner,
     signers::{AwsSignerError, LocalWallet, Signer},
 };
 

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

@@ -5,7 +5,7 @@ use ethers::core::types::H256;
 use crate::{
     accumulator::merkle::Proof,
     traits::{AbacusCommon, ChainCommunicationError, TxOutcome},
-    AbacusMessage, MessageStatus,
+    AbacusMessage, MessageStatus, SignedCheckpoint,
 };
 
 /// Interface for on-chain inboxes
@@ -33,4 +33,11 @@ pub trait Inbox: AbacusCommon + Send + Sync + std::fmt::Debug {
 
     /// Fetch the status of a message
     async fn message_status(&self, leaf: H256) -> Result<MessageStatus, ChainCommunicationError>;
+
+    /// Submit a signed checkpoint for inclusion
+    /// Mocks already have a function called checkpoint
+    async fn submit_checkpoint(
+        &self,
+        signed_checkpoint: &SignedCheckpoint,
+    ) -> Result<TxOutcome, ChainCommunicationError>;
 }

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

@@ -16,7 +16,7 @@ use ethers::{
 };
 use std::error::Error as StdError;
 
-use crate::{db::DbError, utils::home_domain_hash, AbacusError, SignedUpdate};
+use crate::{db::DbError, utils::home_domain_hash, AbacusError, Checkpoint, SignedUpdate};
 
 pub use common::*;
 pub use encode::*;
@@ -164,4 +164,10 @@ pub trait AbacusCommon: Sync + Send + std::fmt::Debug {
 
     /// Fetch the current root.
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError>;
+
+    /// Return the latest checkpointed root and its index.
+    async fn latest_checkpoint(
+        &self,
+        lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError>;
 }

rust/abacus-core/src/types/checkpoint.rs

@@ -16,7 +16,6 @@ pub struct Checkpoint {
     /// The checkpointed root
     pub root: H256,
     /// The index of the checkpoint
-    /// TODO: change to larger than 32 bits
     pub index: u32,
 }
 
@@ -67,12 +66,14 @@ impl Decode for Checkpoint {
 
 impl Checkpoint {
     fn signing_hash(&self) -> H256 {
+        let buffer = [0u8; 28];
         // sign:
-        // domain(home_domain) || previous_root || new_root
+        // domain(home_domain) || root || index (as u256)
         H256::from_slice(
             Keccak256::new()
                 .chain(home_domain_hash(self.outbox_domain))
                 .chain(self.root)
+                .chain(buffer)
                 .chain(self.index.to_be_bytes())
                 .finalize()
                 .as_slice(),

rust/abacus-test/src/mocks/inbox.rs

@@ -24,6 +24,11 @@ mock! {
             proof: &Proof,
         ) -> Result<TxOutcome, ChainCommunicationError> {}
 
+        pub fn _checkpoint(
+            &self,
+            signed_checkpoint: &SignedCheckpoint,
+        ) -> Result<TxOutcome, ChainCommunicationError> {}
+
         // Common
         pub fn _name(&self) -> &str {}
 
@@ -34,6 +39,8 @@ mock! {
         pub fn _checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {}
 
         pub fn _message_status(&self, leaf: H256) -> Result<MessageStatus, ChainCommunicationError> {}
+
+        pub fn _latest_checkpoint(&self, maybe_lag: Option<u64>) -> Result<Checkpoint, ChainCommunicationError> {}
     }
 }
 
@@ -68,6 +75,13 @@ impl Inbox for MockInboxContract {
     async fn message_status(&self, leaf: H256) -> Result<MessageStatus, ChainCommunicationError> {
         self._message_status(leaf)
     }
+
+    async fn submit_checkpoint(
+        &self,
+        signed_checkpoint: &SignedCheckpoint,
+    ) -> Result<TxOutcome, ChainCommunicationError> {
+        self._checkpoint(signed_checkpoint)
+    }
 }
 
 #[async_trait]
@@ -91,4 +105,11 @@ impl AbacusCommon for MockInboxContract {
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {
         self._checkpointed_root()
     }
+
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        self._latest_checkpoint(maybe_lag)
+    }
 }

rust/abacus-test/src/mocks/outbox.rs

@@ -45,6 +45,8 @@ mock! {
         pub fn _state(&self) -> Result<State, ChainCommunicationError> {}
 
         pub fn _checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {}
+
+        pub fn _latest_checkpoint(&self, maybe_lag: Option<u64>) -> Result<Checkpoint, ChainCommunicationError> {}
     }
 }
 
@@ -90,4 +92,11 @@ impl AbacusCommon for MockOutboxContract {
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {
         self._checkpointed_root()
     }
+
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        self._latest_checkpoint(maybe_lag)
+    }
 }

rust/agents/validator/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "validator"
+version = "0.1.0"
+authors = ["Abacus Team"]
+edition = "2021"
+
+[dependencies]
+tokio = { version = "1.0.1", features = ["rt", "macros"] }
+config = "0.10"
+serde = "1.0.120"
+serde_json = { version = "1.0.61", default-features = false }
+log = "0.4.13"
+ethers = { git = "https://github.com/gakonst/ethers-rs", branch = "master" }
+thiserror = { version = "1.0.22", default-features = false }
+async-trait = { version = "0.1.42", default-features = false }
+futures-util = "0.3.12"
+color-eyre = "0.5.0"
+tracing = "0.1.22"
+tracing-futures = "0.2.4"
+tracing-subscriber = "0.2.15"
+rocksdb = { git = "https://github.com/rust-rocksdb/rust-rocksdb" }
+
+abacus-core = { path = "../../abacus-core" }
+abacus-base = { path = "../../abacus-base" }
+abacus-ethereum = { path = "../../chains/abacus-ethereum" }
+paste = "1.0.5"
+
+[dev-dependencies]
+tokio-test = "0.4.0"
+abacus-test = { path = "../../abacus-test" }
+prometheus = "0.12"

rust/agents/validator/src/main.rs

@@ -0,0 +1,40 @@
+//! The checkpointer observes the Outbox contract and calls checkpoint.
+
+#![forbid(unsafe_code)]
+#![warn(missing_docs)]
+#![warn(unused_extern_crates)]
+
+mod settings;
+mod submit;
+mod validator;
+
+use color_eyre::Result;
+
+use abacus_base::Agent;
+
+use crate::{settings::ValidatorSettings as Settings, validator::Validator};
+
+async fn _main() -> Result<()> {
+    color_eyre::install()?;
+    let settings = Settings::new()?;
+
+    let agent = Validator::from_settings(settings).await?;
+
+    agent
+        .as_ref()
+        .settings
+        .tracing
+        .start_tracing(agent.metrics().span_duration())?;
+    let _ = agent.metrics().run_http_server();
+
+    agent.run().await??;
+    Ok(())
+}
+
+fn main() -> Result<()> {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(_main())
+}

rust/agents/validator/src/settings.rs

@@ -0,0 +1,12 @@
+//! Configuration
+
+use abacus_base::decl_settings;
+
+decl_settings!(Validator {
+    /// The validator attestation signer
+    validator: abacus_base::SignerConf,
+    /// The reorg_period in blocks
+    reorgperiod: String,
+    /// How frequently to check for new checkpoints
+    interval: String,
+});

rust/agents/validator/src/submit.rs

@@ -0,0 +1,57 @@
+use std::sync::Arc;
+
+use abacus_base::{CachingOutbox, CheckpointSyncer, LocalStorage};
+use abacus_core::{AbacusCommon, Signers};
+use std::time::Duration;
+
+use color_eyre::Result;
+use tokio::{task::JoinHandle, time::sleep};
+use tracing::{info, info_span, instrument::Instrumented, Instrument};
+pub(crate) struct ValidatorSubmitter {
+    interval: u64,
+    reorg_period: u64,
+    signer: Arc<Signers>,
+    outbox: Arc<CachingOutbox>,
+}
+
+impl ValidatorSubmitter {
+    pub(crate) fn new(
+        interval: u64,
+        reorg_period: u64,
+        outbox: Arc<CachingOutbox>,
+        signer: Arc<Signers>,
+    ) -> Self {
+        Self {
+            reorg_period,
+            interval,
+            outbox,
+            signer,
+        }
+    }
+
+    pub(crate) fn spawn(self) -> Instrumented<JoinHandle<Result<()>>> {
+        let span = info_span!("ValidatorSubmitter");
+        let reorg_period = Some(self.reorg_period);
+        tokio::spawn(async move {
+            let starting_checkpoint = self.outbox.latest_checkpoint(reorg_period).await?;
+            let mut current_index = starting_checkpoint.index;
+            loop {
+                sleep(Duration::from_secs(self.interval)).await;
+
+                // Check the current checkpoint
+                let checkpoint = self.outbox.latest_checkpoint(reorg_period).await?;
+
+                if current_index < checkpoint.index {
+                    let signed_checkpoint = checkpoint.sign_with(self.signer.as_ref()).await?;
+
+                    info!(signature = ?signed_checkpoint, signer=?self.signer, "Sign latest checkpoint");
+                    current_index = checkpoint.index;
+
+                    let storage = LocalStorage { path: "/tmp/validatorsignatures".to_string() };
+                    storage.write_checkpoint(signed_checkpoint).await?;
+                }
+            }
+        })
+        .instrument(span)
+    }
+}

rust/agents/validator/src/validator.rs

@@ -0,0 +1,75 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use color_eyre::Result;
+use tokio::task::JoinHandle;
+use tracing::instrument::Instrumented;
+
+use crate::{settings::ValidatorSettings as Settings, submit::ValidatorSubmitter};
+use abacus_base::{AbacusAgentCore, Agent};
+use abacus_core::Signers;
+
+/// An validator agent
+#[derive(Debug)]
+pub struct Validator {
+    signer: Arc<Signers>,
+    reorg_period: u64,
+    interval: u64,
+    pub(crate) core: AbacusAgentCore,
+}
+
+impl AsRef<AbacusAgentCore> for Validator {
+    fn as_ref(&self) -> &AbacusAgentCore {
+        &self.core
+    }
+}
+
+impl Validator {
+    /// Instantiate a new validator
+    pub fn new(signer: Signers, reorg_period: u64, interval: u64, core: AbacusAgentCore) -> Self {
+        Self {
+            signer: Arc::new(signer),
+            reorg_period,
+            interval,
+            core,
+        }
+    }
+}
+
+#[async_trait]
+impl Agent for Validator {
+    const AGENT_NAME: &'static str = "validator";
+
+    type Settings = Settings;
+
+    async fn from_settings(settings: Self::Settings) -> Result<Self>
+    where
+        Self: Sized,
+    {
+        let signer = settings.validator.try_into_signer().await?;
+        let reorg_period = settings.reorgperiod.parse().expect("invalid uint");
+        let interval = settings.interval.parse().expect("invalid uint");
+        let core = settings
+            .as_ref()
+            .try_into_abacus_core(Self::AGENT_NAME)
+            .await?;
+        Ok(Self::new(signer, reorg_period, interval, core))
+    }
+}
+
+impl Validator {
+    pub fn run(&self) -> Instrumented<JoinHandle<Result<()>>> {
+        let outbox = self.outbox();
+        let submit = ValidatorSubmitter::new(
+            self.interval,
+            self.reorg_period,
+            outbox,
+            self.signer.clone(),
+        );
+
+        self.run_all(vec![submit.spawn()])
+    }
+}
+
+#[cfg(test)]
+mod test {}

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

@@ -97,7 +97,7 @@ where
         Ok(events
             .iter()
             .map(|event| {
-                let signature = Signature::try_from(event.0.signature.as_slice())
+                let signature = Signature::try_from(event.0.signature.as_ref())
                     .expect("chain accepted invalid signature");
 
                 let update = Update {
@@ -139,7 +139,7 @@ where
             .map(|f| RawCommittedMessage {
                 leaf_index: f.leaf_index.as_u32(),
                 committed_root: f.committed_root.into(),
-                message: f.message,
+                message: f.message.to_vec(),
             })
             .collect())
     }
@@ -226,7 +226,7 @@ where
         let tx = self.contract.update(
             update.update.previous_root.to_fixed_bytes(),
             update.update.new_root.to_fixed_bytes(),
-            update.signature.to_vec(),
+            update.signature.to_vec().into(),
         );
 
         Ok(report_tx!(tx).into())
@@ -243,8 +243,8 @@ where
                 double.0.update.new_root.to_fixed_bytes(),
                 double.1.update.new_root.to_fixed_bytes(),
             ],
-            double.0.signature.to_vec(),
-            double.1.signature.to_vec(),
+            double.0.signature.to_vec().into(),
+            double.1.signature.to_vec().into(),
         );
         let response = report_tx!(tx);
 
@@ -271,7 +271,7 @@ where
         let tx = self.contract.dispatch(
             message.destination,
             message.recipient.to_fixed_bytes(),
-            message.body.clone(),
+            message.body.clone().into(),
         );
 
         Ok(report_tx!(tx).into())
@@ -289,7 +289,7 @@ where
         let tx = self.contract.improper_update(
             update.update.previous_root.to_fixed_bytes(),
             update.update.new_root.to_fixed_bytes(),
-            update.signature.to_vec(),
+            update.signature.to_vec().into(),
         );
 
         Ok(report_tx!(tx).into())

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

@@ -103,7 +103,7 @@ where
             ordering
         });
 
-        let outbox_domain = self.contract.local_domain().call().await?;
+        let outbox_domain = self.contract.remote_domain().call().await?;
 
         Ok(events
             .iter()
@@ -194,6 +194,34 @@ where
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {
         Ok(self.contract.checkpointed_root().call().await?.into())
     }
+
+    #[tracing::instrument(err, skip(self))]
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        // This should probably moved into its own trait
+        let base_call = self.contract.latest_checkpoint();
+        let call_with_lag = match maybe_lag {
+            Some(lag) => {
+                let tip = self
+                    .provider
+                    .get_block_number()
+                    .await
+                    .map_err(|x| ChainCommunicationError::CustomError(Box::new(x)))?
+                    .as_u64();
+                base_call.block(if lag > tip { 0 } else { tip - lag })
+            }
+            None => base_call,
+        };
+        let (root, index) = call_with_lag.call().await?;
+        Ok(Checkpoint {
+            // This is inefficient, but latest_checkpoint should never be called
+            outbox_domain: self.remote_domain().await?,
+            root: root.into(),
+            index: index.as_u32(),
+        })
+    }
 }
 
 #[async_trait]
@@ -222,7 +250,7 @@ where
 
     #[tracing::instrument(err)]
     async fn process(&self, message: &AbacusMessage) -> Result<TxOutcome, ChainCommunicationError> {
-        let tx = self.contract.process(message.to_vec());
+        let tx = self.contract.process(message.to_vec().into());
         let gas = tx.estimate_gas().await?.saturating_add(U256::from(100000));
         let gassed = tx.gas(gas);
         Ok(report_tx!(gassed).into())
@@ -241,9 +269,9 @@ where
             .for_each(|(i, elem)| *elem = proof.path[i].to_fixed_bytes());
 
         //
-        let tx = self
-            .contract
-            .prove_and_process(message.to_vec(), sol_proof, proof.index.into());
+        let tx =
+            self.contract
+                .prove_and_process(message.to_vec().into(), sol_proof, proof.index.into());
         let gas = tx.estimate_gas().await?.saturating_add(U256::from(100000));
         let gassed = tx.gas(gas);
         Ok(report_tx!(gassed).into())
@@ -259,4 +287,18 @@ where
             _ => panic!("Bad status from solidity"),
         }
     }
+
+    #[tracing::instrument(err, skip(self), fields(hex_signature = %format!("0x{}", hex::encode(signed_checkpoint.signature.to_vec()))))]
+    async fn submit_checkpoint(
+        &self,
+        signed_checkpoint: &SignedCheckpoint,
+    ) -> Result<TxOutcome, ChainCommunicationError> {
+        let tx = self.contract.checkpoint(
+            signed_checkpoint.checkpoint.root.to_fixed_bytes(),
+            signed_checkpoint.checkpoint.index.into(),
+            signed_checkpoint.signature.to_vec().into(),
+        );
+
+        Ok(report_tx!(tx).into())
+    }
 }

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

@@ -139,7 +139,7 @@ where
             .map(|f| RawCommittedMessage {
                 leaf_index: f.leaf_index.as_u32(),
                 committed_root: f.checkpointed_root.into(),
-                message: f.message,
+                message: f.message.to_vec(),
             })
             .collect())
     }
@@ -214,6 +214,33 @@ where
     async fn checkpointed_root(&self) -> Result<H256, ChainCommunicationError> {
         Ok(self.contract.checkpointed_root().call().await?.into())
     }
+
+    #[tracing::instrument(err, skip(self))]
+    async fn latest_checkpoint(
+        &self,
+        maybe_lag: Option<u64>,
+    ) -> Result<Checkpoint, ChainCommunicationError> {
+        // This should probably moved into its own trait
+        let base_call = self.contract.latest_checkpoint();
+        let call_with_lag = match maybe_lag {
+            Some(lag) => {
+                let tip = self
+                    .provider
+                    .get_block_number()
+                    .await
+                    .map_err(|x| ChainCommunicationError::CustomError(Box::new(x)))?
+                    .as_u64();
+                base_call.block(if lag > tip { 0 } else { tip - lag })
+            }
+            None => base_call,
+        };
+        let (root, index) = call_with_lag.call().await?;
+        Ok(Checkpoint {
+            outbox_domain: self.domain,
+            root: root.into(),
+            index: index.as_u32(),
+        })
+    }
 }
 
 #[async_trait]
@@ -231,7 +258,7 @@ where
         let tx = self.contract.dispatch(
             message.destination,
             message.recipient.to_fixed_bytes(),
-            message.body.clone(),
+            message.body.clone().into(),
         );
 
         Ok(report_tx!(tx).into())

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

@@ -99,7 +99,7 @@ where
         Ok(events
             .iter()
             .map(|event| {
-                let signature = Signature::try_from(event.0.signature.as_slice())
+                let signature = Signature::try_from(event.0.signature.as_ref())
                     .expect("chain accepted invalid signature");
 
                 let update = Update {
@@ -200,7 +200,7 @@ where
         let tx = self.contract.update(
             update.update.previous_root.to_fixed_bytes(),
             update.update.new_root.to_fixed_bytes(),
-            update.signature.to_vec(),
+            update.signature.to_vec().into(),
         );
 
         let result = report_tx!(tx);
@@ -218,8 +218,8 @@ where
                 double.0.update.new_root.to_fixed_bytes(),
                 double.1.update.new_root.to_fixed_bytes(),
             ],
-            double.0.signature.to_vec(),
-            double.1.signature.to_vec(),
+            double.0.signature.to_vec().into(),
+            double.1.signature.to_vec().into(),
         );
 
         Ok(report_tx!(tx).into())
@@ -256,7 +256,7 @@ where
 
     #[tracing::instrument(err)]
     async fn process(&self, message: &AbacusMessage) -> Result<TxOutcome, ChainCommunicationError> {
-        let tx = self.contract.process(message.to_vec());
+        let tx = self.contract.process(message.to_vec().into());
         let gas = tx.estimate_gas().await?.saturating_add(U256::from(100000));
         let gassed = tx.gas(gas);
         Ok(report_tx!(gassed).into())
@@ -275,9 +275,9 @@ where
             .for_each(|(i, elem)| *elem = proof.path[i].to_fixed_bytes());
 
         //
-        let tx = self
-            .contract
-            .prove_and_process(message.to_vec(), sol_proof, proof.index.into());
+        let tx =
+            self.contract
+                .prove_and_process(message.to_vec().into(), sol_proof, proof.index.into());
         let gas = tx.estimate_gas().await?.saturating_add(U256::from(100000));
         let gassed = tx.gas(gas);
         Ok(report_tx!(gassed).into())

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

@@ -132,7 +132,7 @@ where
         let tx = self.contract.unenroll_replica(
             signed_failure.notification.home_domain,
             signed_failure.notification.updater.into(),
-            signed_failure.signature.to_vec(),
+            signed_failure.signature.to_vec().into(),
         );
 
         Ok(report_tx!(tx).into())

rust/tools/kms-cli/src/main.rs

@@ -3,7 +3,7 @@ use std::convert::TryFrom;
 use color_eyre::Result;
 
 use ethers::{
-    prelude::{transaction::eip2718::TypedTransaction, Address, TransactionRequest, U256},
+    core::types::{transaction::eip2718::TypedTransaction, Address, TransactionRequest, U256},
     providers::{Http, Middleware, Provider},
     signers::{AwsSigner, Signer},
 };
@@ -160,10 +160,11 @@ async fn _send_tx(signer: &AwsSigner<'_>, opts: &Opts) -> Result<()> {
     // TODO: remove this these ethers is fixed
     typed_tx.set_gas(21000);
     typed_tx.set_gas_price(20_000_000_000u64); // 20 gwei
+    typed_tx.set_chain_id(signer.chain_id());
 
     let sig = signer.sign_transaction(&typed_tx).await?;
 
-    let rlp = typed_tx.rlp_signed(signer.chain_id(), &sig);
+    let rlp = typed_tx.rlp_signed(&sig);
     println!(
         "Tx request details:\n{}",
         serde_json::to_string_pretty(&typed_tx)?