Merge pull request #339 from consensus-shipyard/relayer-cli

Relayer cli

ipc/cli/src/commands/checkpoint/mod.rs

@@ -3,10 +3,12 @@
 use crate::commands::checkpoint::list_checkpoints::{
     ListBottomUpCheckpoints, ListBottomUpCheckpointsArgs,
 };
+use crate::commands::checkpoint::relayer::{BottomUpRelayer, BottomUpRelayerArgs};
 use crate::{CommandLineHandler, GlobalArguments};
 use clap::{Args, Subcommand};
 
 mod list_checkpoints;
+mod relayer;
 
 #[derive(Debug, Args)]
 #[command(name = "checkpoint", about = "checkpoint related commands")]
@@ -20,11 +22,13 @@ impl CheckpointCommandsArgs {
     pub async fn handle(&self, global: &GlobalArguments) -> anyhow::Result<()> {
         match &self.command {
             Commands::ListBottomup(args) => ListBottomUpCheckpoints::handle(global, args).await,
+            Commands::Relayer(args) => BottomUpRelayer::handle(global, args).await,
         }
     }
 }
 
 #[derive(Debug, Subcommand)]
 pub(crate) enum Commands {
     ListBottomup(ListBottomUpCheckpointsArgs),
+    Relayer(BottomUpRelayerArgs),
 }

ipc/cli/src/commands/checkpoint/relayer.rs

@@ -0,0 +1,79 @@
+// Copyright 2022-2023 Protocol Labs
+// SPDX-License-Identifier: MIT
+
+use crate::commands::get_subnet_config;
+use crate::{CommandLineHandler, GlobalArguments};
+use anyhow::anyhow;
+use async_trait::async_trait;
+use clap::Args;
+use fvm_shared::address::Address;
+use ipc_identity::EvmKeyStore;
+use ipc_provider::checkpoint::BottomUpCheckpointManager;
+use ipc_provider::new_evm_keystore_from_path;
+use ipc_sdk::subnet_id::SubnetID;
+use std::str::FromStr;
+use std::sync::{Arc, RwLock};
+use std::time::Duration;
+
+const DEFAULT_POLLING_INTERVAL: u64 = 15;
+
+/// The command to run the bottom up relayer in the background.
+pub(crate) struct BottomUpRelayer;
+
+#[async_trait]
+impl CommandLineHandler for BottomUpRelayer {
+    type Arguments = BottomUpRelayerArgs;
+
+    async fn handle(global: &GlobalArguments, arguments: &Self::Arguments) -> anyhow::Result<()> {
+        log::debug!("start bottom up relayer with args: {:?}", arguments);
+
+        let config_path = global.config_path();
+
+        let mut keystore = new_evm_keystore_from_path(&config_path)?;
+        let submitter = match (arguments.submitter.as_ref(), keystore.get_default()?) {
+            (Some(submitter), _) => Address::from_str(submitter)?,
+            (None, Some(addr)) => {
+                log::info!("using default address: {addr:?}");
+                Address::try_from(addr)?
+            }
+            _ => {
+                return Err(anyhow!("no submitter address provided"));
+            }
+        };
+
+        let subnet = SubnetID::from_str(&arguments.subnet)?;
+        let parent = subnet
+            .parent()
+            .ok_or_else(|| anyhow!("root does not have parent"))?;
+
+        let child = get_subnet_config(&config_path, &subnet)?;
+        let parent = get_subnet_config(&config_path, &parent)?;
+
+        let manager = BottomUpCheckpointManager::new_evm_manager(
+            parent.clone(),
+            child.clone(),
+            Arc::new(RwLock::new(keystore)),
+        )
+        .await?;
+
+        let interval = Duration::from_secs(
+            arguments
+                .checkpoint_interval_sec
+                .unwrap_or(DEFAULT_POLLING_INTERVAL),
+        );
+        manager.run(submitter, interval).await;
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Args)]
+#[command(about = "Start the bottom up relayer daemon")]
+pub(crate) struct BottomUpRelayerArgs {
+    #[arg(long, short, help = "The subnet id of the checkpointing subnet")]
+    pub subnet: String,
+    #[arg(long, short, help = "The number of seconds to submit checkpoint")]
+    pub checkpoint_interval_sec: Option<u64>,
+    #[arg(long, short, help = "The hex encoded address of the submitter")]
+    pub submitter: Option<String>,
+}

ipc/cli/src/commands/mod.rs

@@ -14,15 +14,18 @@ use crate::commands::checkpoint::CheckpointCommandsArgs;
 use crate::commands::crossmsg::CrossMsgsCommandsArgs;
 use crate::commands::util::UtilCommandsArgs;
 use crate::GlobalArguments;
-use anyhow::{Context, Result};
+use anyhow::{anyhow, Context, Result};
 
 use clap::{Command, CommandFactory, Parser, Subcommand};
 use clap_complete::{generate, Generator, Shell};
 use fvm_shared::econ::TokenAmount;
 use ipc_sdk::ethers_address_to_fil_address;
 
+use ipc_provider::config::{Config, Subnet};
+use ipc_sdk::subnet_id::SubnetID;
 use std::fmt::Debug;
 use std::io;
+use std::path::Path;
 use std::str::FromStr;
 
 use crate::commands::config::ConfigCommandsArgs;
@@ -146,6 +149,19 @@ pub(crate) fn require_fil_addr_from_str(s: &str) -> anyhow::Result<fvm_shared::a
     Ok(addr)
 }
 
+/// Get the subnet configuration from the config path
+pub(crate) fn get_subnet_config(
+    config_path: impl AsRef<Path>,
+    subnet: &SubnetID,
+) -> Result<Subnet> {
+    let config = Config::from_file(&config_path)?;
+    Ok(config
+        .subnets
+        .get(subnet)
+        .ok_or_else(|| anyhow!("{subnet} is not configured"))?
+        .clone())
+}
+
 #[cfg(test)]
 mod tests {
     use crate::f64_to_token_amount;

ipc/identity/src/evm/mod.rs

@@ -91,7 +91,7 @@ pub fn random_eth_key_info() -> KeyInfo {
 }
 
 #[cfg(feature = "with-ethers")]
-#[derive(Clone, Eq, Hash, PartialEq, Default)]
+#[derive(Debug, Clone, Eq, Hash, PartialEq, Default)]
 pub struct EthKeyAddress {
     inner: ethers::types::Address,
 }

ipc/provider/src/checkpoint.rs

@@ -20,6 +20,9 @@ pub struct CheckpointConfig {
     period: ChainEpoch,
 }
 
+/// Manages the submission of bottom up checkpoint. It checks if the submitter has already
+/// submitted in the `last_checkpoint_height`, if not, it will submit the checkpoint at that height.
+/// Then it will submit at the next submission height for the new checkpoint.
 pub struct BottomUpCheckpointManager<T> {
     metadata: CheckpointConfig,
     parent_handler: T,
@@ -88,21 +91,63 @@ impl<T: BottomUpCheckpointRelayer + Send + Sync + 'static> BottomUpCheckpointMan
         self.metadata.period
     }
 
-    /// Run the bottom up checkpoint submission daemon in the background
-    pub fn run(self, validator: Address, submission_interval: Duration) {
-        tokio::spawn(async move {
-            loop {
-                if let Err(e) = self.submit_checkpoint(&validator).await {
-                    log::error!("cannot submit checkpoint for validator: {validator} due to {e}");
-                }
-
-                tokio::time::sleep(submission_interval).await;
+    /// Run the bottom up checkpoint submission daemon in the foreground
+    pub async fn run(self, submitter: Address, submission_interval: Duration) {
+        loop {
+            if let Err(e) = self.submit_checkpoint(&submitter).await {
+                log::error!("cannot submit checkpoint for submitter: {submitter} due to {e}");
             }
-        });
+
+            tokio::time::sleep(submission_interval).await;
+        }
+    }
+
+    /// Submit the checkpoint from the target submitter address
+    pub async fn submit_checkpoint(&self, submitter: &Address) -> Result<()> {
+        self.submit_last_epoch(submitter).await?;
+        self.submit_next_epoch(submitter).await
+    }
+
+    /// Derive the next submission checkpoint height
+    async fn next_submission_height(&self) -> Result<ChainEpoch> {
+        let last_checkpoint_epoch = self
+            .parent_handler
+            .last_bottom_up_checkpoint_height(&self.metadata.child.id)
+            .await
+            .map_err(|e| {
+                anyhow!("cannot obtain the last bottom up checkpoint height due to: {e:}")
+            })?;
+        Ok(last_checkpoint_epoch + self.checkpoint_period())
+    }
+
+    /// Checks if the relayer has already submitted at the `last_checkpoint_height`, if not it submits it.
+    async fn submit_last_epoch(&self, submitter: &Address) -> Result<()> {
+        let subnet = &self.metadata.child.id;
+        if self
+            .child_handler
+            .has_submitted_in_last_checkpoint_height(subnet, submitter)
+            .await?
+        {
+            return Ok(());
+        }
+
+        let height = self
+            .child_handler
+            .last_bottom_up_checkpoint_height(subnet)
+            .await?;
+        let bundle = self.child_handler.checkpoint_bundle_at(height).await?;
+        log::debug!("bottom up bundle: {bundle:?}");
+
+        self.parent_handler
+            .submit_checkpoint(submitter, bundle)
+            .await
+            .map_err(|e| anyhow!("cannot submit bottom up checkpoint due to: {e:}"))?;
+
+        Ok(())
     }
 
-    /// Submit the checkpoint from the target validator address
-    pub async fn submit_checkpoint(&self, validator: &Address) -> Result<()> {
+    /// Checks if the relayer has already submitted at the next submission epoch, if not it submits it.
+    async fn submit_next_epoch(&self, submitter: &Address) -> Result<()> {
         let next_submission_height = self.next_submission_height().await?;
         let current_height = self.child_handler.current_epoch().await?;
 
@@ -117,21 +162,10 @@ impl<T: BottomUpCheckpointRelayer + Send + Sync + 'static> BottomUpCheckpointMan
         log::debug!("bottom up bundle: {bundle:?}");
 
         self.parent_handler
-            .submit_checkpoint(validator, bundle)
+            .submit_checkpoint(submitter, bundle)
             .await
             .map_err(|e| anyhow!("cannot submit bottom up checkpoint due to: {e:}"))?;
 
         Ok(())
     }
-
-    async fn next_submission_height(&self) -> Result<ChainEpoch> {
-        let last_checkpoint_epoch = self
-            .parent_handler
-            .last_bottom_up_checkpoint_height(&self.metadata.child.id)
-            .await
-            .map_err(|e| {
-                anyhow!("cannot obtain the last bottom up checkpoint height due to: {e:}")
-            })?;
-        Ok(last_checkpoint_epoch + self.checkpoint_period())
-    }
 }

ipc/provider/src/manager/evm/manager.rs

@@ -841,6 +841,23 @@ impl BottomUpCheckpointRelayer for EthSubnetManager {
         Ok(epoch as ChainEpoch)
     }
 
+    async fn has_submitted_in_last_checkpoint_height(
+        &self,
+        subnet_id: &SubnetID,
+        submitter: &Address,
+    ) -> Result<bool> {
+        let address = contract_address_from_subnet(subnet_id)?;
+        let contract = subnet_actor_getter_facet::SubnetActorGetterFacet::new(
+            address,
+            Arc::new(self.ipc_contract_info.provider.clone()),
+        );
+        let addr = payload_to_evm_address(submitter.payload())?;
+        Ok(contract
+            .has_submitted_in_last_bottom_up_checkpoint_height(addr)
+            .call()
+            .await?)
+    }
+
     async fn checkpoint_period(&self, subnet_id: &SubnetID) -> anyhow::Result<ChainEpoch> {
         let address = contract_address_from_subnet(subnet_id)?;
         let contract = subnet_actor_getter_facet::SubnetActorGetterFacet::new(

ipc/provider/src/manager/subnet.rs

@@ -165,9 +165,15 @@ pub trait BottomUpCheckpointRelayer: Send + Sync {
     ) -> Result<()>;
     /// The last confirmed/submitted checkpoint height.
     async fn last_bottom_up_checkpoint_height(&self, subnet_id: &SubnetID) -> Result<ChainEpoch>;
+    /// Check if the submitter has already submitted in the `last_bottom_up_checkpoint_height`
+    async fn has_submitted_in_last_checkpoint_height(
+        &self,
+        subnet_id: &SubnetID,
+        submitter: &Address,
+    ) -> Result<bool>;
     /// Get the checkpoint period, i.e the number of blocks to submit bottom up checkpoints.
     async fn checkpoint_period(&self, subnet_id: &SubnetID) -> Result<ChainEpoch>;
-    /// Get the checkpoint at a specific height. If it does not exist, it will through error.
+    /// Get the checkpoint bundle at a specific height. If it does not exist, it will through error.
     async fn checkpoint_bundle_at(&self, height: ChainEpoch) -> Result<BottomUpCheckpointBundle>;
     /// Get the current epoch in the current subnet
     async fn current_epoch(&self) -> Result<ChainEpoch>;

ipc/sdk/src/checkpoint.rs

@@ -27,6 +27,7 @@ lazy_static! {
 
 pub type Signature = Vec<u8>;
 
+/// The collection of items for the bottom up checkpoint submission
 #[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
 pub struct BottomUpCheckpointBundle {
     pub checkpoint: BottomUpCheckpoint,