feat: checkpoint fallbacks (#120)

* 🏗️ checkpoint fallback initial impl

* 🚧 checkpoint fallbacks

* ✅ checkpoint fallbacks

* ⚙️ fix result types

* ♻️ checkpoints refactoring

* 🔨 import nits

* 🚀 graceful checkpoint fallbacks

* ⚡ parallel checkpoint fallback service fetching using async tokio tasks

* 📝 readme touchups

Cargo.lock

@@ -597,10 +597,16 @@ dependencies = [
  "ethers",
  "eyre",
  "figment",
+ "futures",
  "hex",
+ "log",
+ "reqwest",
  "serde",
+ "serde_yaml",
  "ssz-rs",
+ "strum",
  "thiserror",
+ "tokio",
 ]
 
 [[package]]
@@ -2900,9 +2906,9 @@ dependencies = [
 
 [[package]]
 name = "reqwest"
-version = "0.11.12"
+version = "0.11.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "431949c384f4e2ae07605ccaa56d1d9d2ecdb5cadd4f9577ccfab29f2e5149fc"
+checksum = "68cc60575865c7831548863cc02356512e3f1dc2f3f82cb837d7fc4cc8f3c97c"
 dependencies = [
  "base64 0.13.1",
  "bytes",
@@ -3380,6 +3386,19 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_yaml"
+version = "0.9.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d232d893b10de3eb7258ff01974d6ee20663d8e833263c99409d4b13a0209da"
+dependencies = [
+ "indexmap",
+ "itoa",
+ "ryu",
+ "serde",
+ "unsafe-libyaml",
+]
+
 [[package]]
 name = "sha-1"
 version = "0.9.8"
@@ -3980,6 +3999,12 @@ version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f962df74c8c05a667b5ee8bcf162993134c104e96440b663c8daa176dc772d8c"
 
+[[package]]
+name = "unsafe-libyaml"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1e5fa573d8ac5f1a856f8d7be41d390ee973daf97c806b2c1a465e4e1406e68"
+
 [[package]]
 name = "untrusted"
 version = "0.7.1"

README.md

@@ -1,4 +1,5 @@
 ## Helios
+
 [![build](https://github.com/a16z/helios/actions/workflows/test.yml/badge.svg)](https://github.com/a16z/helios/actions/workflows/test.yml) [![license: MIT](https://img.shields.io/badge/license-MIT-brightgreen.svg)](https://opensource.org/licenses/MIT) [![chat](https://img.shields.io/badge/chat-telegram-blue)](https://t.me/+IntDY_gZJSRkNTJj)
 
 Helios is a fully trustless, efficient, and portable Ethereum light client written in Rust.
@@ -8,14 +9,19 @@ Helios converts an untrusted centralized RPC endpoint into a safe unmanipulable
 The entire size of Helios's binary is 13Mb and should be easy to compile into WebAssembly. This makes it a perfect target to embed directly inside wallets and dapps.
 
 ## Installing
+
 First install `heliosup`, Helios's installer:
+
 ```
 curl https://raw.githubusercontent.com/a16z/helios/master/heliosup/install | bash
 ```
+
 To install Helios, run `heliosup`.
 
 ## Usage
+
 To run Helios, run the below command, replacing `$ETH_RPC_URL` with an RPC provider URL such as Alchemy or Infura:
+
 ```
 helios --execution-rpc $ETH_RPC_URL
 ```
@@ -25,9 +31,11 @@ helios --execution-rpc $ETH_RPC_URL
 Helios will now run a local RPC server at `http://127.0.0.1:8545`.
 
 ### Warning
+
 Helios is still experimental software. While we hope you try it out, we do not suggest adding it as your main RPC in wallets yet. Sending high-value transactions from a wallet connected to Helios is discouraged.
 
 ### Additional Options
+
 `--consensus-rpc` or `-c` can be used to set a custom consensus layer rpc endpoint. This must be a consenus node that supports the light client beaconchain api. We recommend using Nimbus for this. If no consensus rpc is supplied, it defaults to `https://www.lightclientdata.org` which is run by us.
 
 `--checkpoint` or `-w` can be used to set a custom weak subjectivity checkpoint. This must be equal the first beacon blockhash of an epoch. Weak subjectivity checkpoints are the root of trust in the system. If this is set to a malicious value, an attacker can cause the client to sync to the wrong chain. Helios sets a default value initially, then caches the most recent finalized block it has seen for later use.
@@ -38,8 +46,25 @@ Helios is still experimental software. While we hope you try it out, we do not s
 
 `--data-dir` or `-d` sets the directory that Helios should use to store cached weak subjectivity checkpoints in. Each network only stores the latest checkpoint, which is just 32 bytes.
 
+`--fallback` or `-f` sets the checkpoint fallback url (a string). This is only used if the checkpoint provided by the `--checkpoint` flag is too outdated for Helios to use to sync.
+If none is provided and the `--load-external-fallback` flag is not set, Helios will error.
+For example, you can specify the fallback like so: `helios --fallback "https://sync-mainnet.beaconcha.in"` (or using shorthand like so: `helios -f "https://sync-mainnet.beaconcha.in"`)
+
+`--load-external-fallback` or `-l` enables weak subjectivity checkpoint fallback (no value needed).
+For example, say you set a checkpoint value that is too outdated and Helios cannot sync to it.
+If this flag is set, Helios will query all network apis in the community-maintained list
+at [ethpandaops/checkpoint-synz-health-checks](https://github.com/ethpandaops/checkpoint-sync-health-checks/blob/master/_data/endpoints.yaml) for their latest slots.
+The list of slots is filtered for healthy apis and the most frequent checkpoint occuring in the latest epoch will be returned.
+Note: this is a community-maintained list and thus no security guarantees are provided. Use this is a last resort if your checkpoint passed into `--checkpoint` fails.
+This is not recommened as malicious checkpoints can be returned from the listed apis, even if they are considered _healthy_.
+This can be run like so: `helios --load-external-fallback` (or `helios -l` with the shorthand).
+
+`--help` or `-h` prints the help message.
+
 ### Configuration Files
+
 All configuration options can be set on a per-network level in `~/.helios/helios.toml`. Here is an example config file:
+
 ```toml
 [mainnet]
 consensus_rpc = "https://www.lightclientdata.org"
@@ -53,6 +78,7 @@ checkpoint = "0xb5c375696913865d7c0e166d87bc7c772b6210dc9edf149f4c7ddc6da0dd4495
 ```
 
 ### Using Helios as a Library
+
 Helios can be imported into any Rust project. Helios requires the Rust nightly toolchain to compile.
 
 ```rust
@@ -86,11 +112,39 @@ async fn main() -> Result<()> {
 }
 ```
 
+Below we demonstrate fetching checkpoints from the community-maintained list of checkpoint sync apis maintained by [ethPandaOps](https://github.com/ethpandaops/checkpoint-sync-health-checks/blob/master/_data/endpoints.yaml).
+
+> **Warning**
+>
+> This is a community-maintained list and thus no security guarantees are provided. Attacks on your light client can occur if malicious checkpoints are set in the list. Please use the explicit `checkpoint` flag, environment variable, or config setting with an updated, and verified checkpoint.
+
+```rust
+use eyre::Result;
+use helios::config::{checkpoints, networks};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Construct the checkpoint fallback services
+    let cf = checkpoints::CheckpointFallback::new().build().await.unwrap();
+
+    // Fetch the latest goerli checkpoint
+    let goerli_checkpoint = cf.fetch_latest_checkpoint(&networks::Network::GOERLI).await.unwrap();
+    println!("Fetched latest goerli checkpoint: {}", goerli_checkpoint);
+
+    // Fetch the latest mainnet checkpoint
+    let mainnet_checkpoint = cf.fetch_latest_checkpoint(&networks::Network::MAINNET).await.unwrap();
+    println!("Fetched latest mainnet checkpoint: {}", mainnet_checkpoint);
+}
+```
+
 ## Contributing
+
 All contributions to Helios are welcome. Before opening a PR, please submit an issue detailing the bug or feature. When opening a PR, please ensure that your contribution builds on the nightly rust toolchain, has been linted with `cargo fmt`, and contains tests when applicable.
 
 ## Telegram
+
 If you are having trouble with Helios or are considering contributing, feel free to join our telegram [here](https://t.me/+IntDY_gZJSRkNTJj).
 
 ## Disclaimer
+
 _This code is being provided as is. No guarantee, representation or warranty is being made, express or implied, as to the safety or correctness of the code. It has not been audited and as such there can be no assurance it will work as intended, and users may experience delays, failures, errors, omissions or loss of transmitted information. Nothing in this repo should be construed as investment advice or legal advice for any particular facts or circumstances and is not meant to replace competent counsel. It is strongly advised for you to contact a reputable attorney in your jurisdiction for any questions or concerns with respect thereto. a16z is not liable for any use of the foregoing, and users should proceed with caution and use at their own risk. See a16z.com/disclosures for more info._

cli/src/main.rs

@@ -85,6 +85,10 @@ struct Cli {
     consensus_rpc: Option<String>,
     #[clap(short, long, env)]
     data_dir: Option<String>,
+    #[clap(short = 'f', long, env)]
+    fallback: Option<String>,
+    #[clap(short = 'l', long, env)]
+    load_external_fallback: bool,
 }
 
 impl Cli {
@@ -100,6 +104,8 @@ impl Cli {
             consensus_rpc: self.consensus_rpc.clone(),
             data_dir: self.get_data_dir(),
             rpc_port: self.rpc_port,
+            fallback: self.fallback.clone(),
+            load_external_fallback: self.load_external_fallback,
         }
     }
 

client/src/client.rs

@@ -7,8 +7,8 @@ use ethers::types::{Filter, Log, Transaction, TransactionReceipt, H256};
 use eyre::{eyre, Result};
 
 use common::types::BlockTag;
-use config::Config;
+use config::{CheckpointFallback, Config};
-use consensus::types::Header;
+use consensus::{types::Header, ConsensusClient};
 use execution::types::{CallOpts, ExecutionBlock};
 use log::{info, warn};
 use tokio::spawn;
@@ -19,27 +19,6 @@ use crate::database::{Database, FileDB};
 use crate::node::Node;
 use crate::rpc::Rpc;
 
-pub struct Client<DB: Database> {
-    node: Arc<RwLock<Node>>,
-    rpc: Option<Rpc>,
-    db: Option<DB>,
-}
-
-impl Client<FileDB> {
-    fn new(config: Config) -> Result<Self> {
-        let config = Arc::new(config);
-        let node = Node::new(config.clone())?;
-        let node = Arc::new(RwLock::new(node));
-
-        let rpc = config.rpc_port.map(|port| Rpc::new(node.clone(), port));
-
-        let data_dir = config.data_dir.clone();
-        let db = data_dir.map(FileDB::new);
-
-        Ok(Client { node, rpc, db })
-    }
-}
-
 #[derive(Default)]
 pub struct ClientBuilder {
     network: Option<Network>,
@@ -49,6 +28,8 @@ pub struct ClientBuilder {
     rpc_port: Option<u16>,
     data_dir: Option<PathBuf>,
     config: Option<Config>,
+    fallback: Option<String>,
+    load_external_fallback: bool,
 }
 
 impl ClientBuilder {
@@ -92,6 +73,16 @@ impl ClientBuilder {
         self
     }
 
+    pub fn fallback(mut self, fallback: &str) -> Self {
+        self.fallback = Some(fallback.to_string());
+        self
+    }
+
+    pub fn load_external_fallback(mut self) -> Self {
+        self.load_external_fallback = true;
+        self
+    }
+
     pub fn build(self) -> Result<Client<FileDB>> {
         let base_config = if let Some(network) = self.network {
             network.to_base_config()
@@ -143,6 +134,20 @@ impl ClientBuilder {
             None
         };
 
+        let fallback = if self.fallback.is_some() {
+            self.fallback
+        } else if let Some(config) = &self.config {
+            config.fallback.clone()
+        } else {
+            None
+        };
+
+        let load_external_fallback = if let Some(config) = &self.config {
+            self.load_external_fallback || config.load_external_fallback
+        } else {
+            self.load_external_fallback
+        };
+
         let config = Config {
             consensus_rpc,
             execution_rpc,
@@ -152,21 +157,60 @@ impl ClientBuilder {
             chain: base_config.chain,
             forks: base_config.forks,
             max_checkpoint_age: base_config.max_checkpoint_age,
+            fallback,
+            load_external_fallback,
         };
 
         Client::new(config)
     }
 }
 
+pub struct Client<DB: Database> {
+    node: Arc<RwLock<Node>>,
+    rpc: Option<Rpc>,
+    db: Option<DB>,
+    fallback: Option<String>,
+    load_external_fallback: bool,
+}
+
+impl Client<FileDB> {
+    fn new(config: Config) -> Result<Self> {
+        let config = Arc::new(config);
+        let node = Node::new(config.clone())?;
+        let node = Arc::new(RwLock::new(node));
+
+        let rpc = config.rpc_port.map(|port| Rpc::new(node.clone(), port));
+
+        let data_dir = config.data_dir.clone();
+        let db = data_dir.map(FileDB::new);
+
+        Ok(Client {
+            node,
+            rpc,
+            db,
+            fallback: config.fallback.clone(),
+            load_external_fallback: config.load_external_fallback,
+        })
+    }
+}
+
 impl<DB: Database> Client<DB> {
     pub async fn start(&mut self) -> Result<()> {
         if let Some(rpc) = &mut self.rpc {
             rpc.start().await?;
         }
 
-        let res = self.node.write().await.sync().await;
+        if self.node.write().await.sync().await.is_err() {
-        if let Err(err) = res {
+            warn!(
-            warn!("consensus error: {}", err);
+                "failed to sync consensus node with checkpoint: 0x{}",
+                hex::encode(&self.node.read().await.config.checkpoint),
+            );
+            let fallback = self.boot_from_fallback().await;
+            if fallback.is_err() && self.load_external_fallback {
+                self.boot_from_external_fallbacks().await?
+            } else if fallback.is_err() {
+                return Err(eyre::eyre!("Checkpoint is too old. Please update your checkpoint. Alternatively, set an explicit checkpoint fallback service url with the `-f` flag or use the configured external fallback services with `-l` (NOT RECOMMENED). See https://github.com/a16z/helios#additional-options for more information."));
+            }
         }
 
         let node = self.node.clone();
@@ -185,6 +229,75 @@ impl<DB: Database> Client<DB> {
         Ok(())
     }
 
+    async fn boot_from_fallback(&self) -> eyre::Result<()> {
+        if let Some(fallback) = &self.fallback {
+            info!(
+                "attempting to load checkpoint from fallback \"{}\"",
+                fallback
+            );
+
+            let checkpoint = CheckpointFallback::fetch_checkpoint_from_api(fallback)
+                .await
+                .map_err(|_| {
+                    eyre::eyre!("Failed to fetch checkpoint from fallback \"{}\"", fallback)
+                })?;
+
+            info!(
+                "external fallbacks responded with checkpoint 0x{:?}",
+                checkpoint
+            );
+
+            // Try to sync again with the new checkpoint by reconstructing the consensus client
+            // We fail fast here since the node is unrecoverable at this point
+            let config = self.node.read().await.config.clone();
+            let consensus =
+                ConsensusClient::new(&config.consensus_rpc, checkpoint.as_bytes(), config.clone())?;
+            self.node.write().await.consensus = consensus;
+            self.node.write().await.sync().await?;
+
+            Ok(())
+        } else {
+            Err(eyre::eyre!("no explicit fallback specified"))
+        }
+    }
+
+    async fn boot_from_external_fallbacks(&self) -> eyre::Result<()> {
+        info!("attempting to fetch checkpoint from external fallbacks...");
+        // Build the list of external checkpoint fallback services
+        let list = CheckpointFallback::new()
+            .build()
+            .await
+            .map_err(|_| eyre::eyre!("Failed to construct external checkpoint sync fallbacks"))?;
+
+        let checkpoint = if self.node.read().await.config.chain.chain_id == 5 {
+            list.fetch_latest_checkpoint(&Network::GOERLI)
+                .await
+                .map_err(|_| {
+                    eyre::eyre!("Failed to fetch latest goerli checkpoint from external fallbacks")
+                })?
+        } else {
+            list.fetch_latest_checkpoint(&Network::MAINNET)
+                .await
+                .map_err(|_| {
+                    eyre::eyre!("Failed to fetch latest mainnet checkpoint from external fallbacks")
+                })?
+        };
+
+        info!(
+            "external fallbacks responded with checkpoint {:?}",
+            checkpoint
+        );
+
+        // Try to sync again with the new checkpoint by reconstructing the consensus client
+        // We fail fast here since the node is unrecoverable at this point
+        let config = self.node.read().await.config.clone();
+        let consensus =
+            ConsensusClient::new(&config.consensus_rpc, checkpoint.as_bytes(), config.clone())?;
+        self.node.write().await.consensus = consensus;
+        self.node.write().await.sync().await?;
+        Ok(())
+    }
+
     pub async fn shutdown(&self) {
         let node = self.node.read().await;
         let checkpoint = if let Some(checkpoint) = node.get_last_checkpoint() {

client/src/node.rs

@@ -20,12 +20,12 @@ use execution::ExecutionClient;
 use crate::errors::NodeError;
 
 pub struct Node {
-    consensus: ConsensusClient<NimbusRpc>,
+    pub consensus: ConsensusClient<NimbusRpc>,
-    execution: Arc<ExecutionClient<HttpRpc>>,
+    pub execution: Arc<ExecutionClient<HttpRpc>>,
-    config: Arc<Config>,
+    pub config: Arc<Config>,
     payloads: BTreeMap<u64, ExecutionPayload>,
     finalized_payloads: BTreeMap<u64, ExecutionPayload>,
-    history_size: usize,
+    pub history_size: usize,
 }
 
 impl Node {

config/Cargo.toml

@@ -5,6 +5,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+tokio = { version = "1", features = ["full"] }
 eyre = "0.6.8"
 serde = { version = "1.0.143", features = ["derive"] }
 hex = "0.4.3"
@@ -12,5 +13,12 @@ ssz-rs = { git = "https://github.com/ralexstokes/ssz-rs", rev = "cb08f18ca919cc1
 ethers = "1.0.0"
 figment = { version = "0.10.7", features = ["toml", "env"] }
 thiserror = "1.0.37"
+log = "0.4.17"
 
 common = { path = "../common" }
+reqwest = "0.11.13"
+serde_yaml = "0.9.14"
+strum = "0.24.1"
+futures = "0.3.25"
+
+

config/src/base.rs

@@ -0,0 +1,19 @@
+use serde::Serialize;
+
+use crate::types::{ChainConfig, Forks};
+use crate::utils::bytes_serialize;
+
+/// The base configuration for a network.
+#[derive(Serialize, Default)]
+pub struct BaseConfig {
+    pub rpc_port: u16,
+    pub consensus_rpc: Option<String>,
+    #[serde(
+        deserialize_with = "bytes_deserialize",
+        serialize_with = "bytes_serialize"
+    )]
+    pub checkpoint: Vec<u8>,
+    pub chain: ChainConfig,
+    pub forks: Forks,
+    pub max_checkpoint_age: u64,
+}

config/src/checkpoints.rs

@@ -0,0 +1,232 @@
+use std::collections::HashMap;
+
+use ethers::types::H256;
+use serde::{Deserialize, Serialize};
+
+use crate::networks;
+
+/// The location where the list of checkpoint services are stored.
+pub const CHECKPOINT_SYNC_SERVICES_LIST: &str = "https://raw.githubusercontent.com/ethpandaops/checkpoint-sync-health-checks/master/_data/endpoints.yaml";
+
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RawSlotResponse {
+    pub data: RawSlotResponseData,
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RawSlotResponseData {
+    pub slots: Vec<Slot>,
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Slot {
+    pub slot: u64,
+    pub block_root: Option<H256>,
+    pub state_root: Option<H256>,
+    pub epoch: u64,
+    pub time: StartEndTime,
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StartEndTime {
+    /// An ISO 8601 formatted UTC timestamp.
+    pub start_time: String,
+    /// An ISO 8601 formatted UTC timestamp.
+    pub end_time: String,
+}
+
+/// A health check for the checkpoint sync service.
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Health {
+    /// If the node is healthy.
+    pub result: bool,
+    /// An [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) UTC timestamp.
+    pub date: String,
+}
+
+/// A checkpoint fallback service.
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CheckpointFallbackService {
+    /// The endpoint for the checkpoint sync service.
+    pub endpoint: String,
+    /// The checkpoint sync service name.
+    pub name: String,
+    /// The service state.
+    pub state: bool,
+    /// If the service is verified.
+    pub verification: bool,
+    /// Contact information for the service maintainers.
+    pub contacts: Option<serde_yaml::Value>,
+    /// Service Notes
+    pub notes: Option<serde_yaml::Value>,
+    /// The service health check.
+    pub health: Vec<Health>,
+}
+
+/// The CheckpointFallback manages checkpoint fallback services.
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CheckpointFallback {
+    /// Services Map
+    pub services: HashMap<networks::Network, Vec<CheckpointFallbackService>>,
+    /// A list of supported networks to build.
+    /// Default: [mainnet, goerli]
+    pub networks: Vec<networks::Network>,
+}
+
+impl CheckpointFallback {
+    /// Constructs a new checkpoint fallback service.
+    pub fn new() -> Self {
+        Self {
+            services: Default::default(),
+            networks: [networks::Network::MAINNET, networks::Network::GOERLI].to_vec(),
+        }
+    }
+
+    /// Build the checkpoint fallback service from the community-maintained list by [ethPandaOps](https://github.com/ethpandaops).
+    ///
+    /// The list is defined in [ethPandaOps/checkpoint-fallback-service](https://github.com/ethpandaops/checkpoint-sync-health-checks/blob/master/_data/endpoints.yaml).
+    pub async fn build(mut self) -> eyre::Result<Self> {
+        // Fetch the services
+        let client = reqwest::Client::new();
+        let res = client.get(CHECKPOINT_SYNC_SERVICES_LIST).send().await?;
+        let yaml = res.text().await?;
+
+        // Parse the yaml content results.
+        let list: serde_yaml::Value = serde_yaml::from_str(&yaml)?;
+
+        // Construct the services mapping from network <> list of services
+        let mut services = HashMap::new();
+        for network in &self.networks {
+            // Try to parse list of checkpoint fallback services
+            let service_list = list
+                .get(network.to_string().to_lowercase())
+                .ok_or_else(|| {
+                    eyre::eyre!(format!("missing {} fallback checkpoint services", network))
+                })?;
+            let parsed: Vec<CheckpointFallbackService> =
+                serde_yaml::from_value(service_list.clone())?;
+            services.insert(*network, parsed);
+        }
+        self.services = services;
+
+        Ok(self)
+    }
+
+    /// Fetch the latest checkpoint from the checkpoint fallback service.
+    pub async fn fetch_latest_checkpoint(
+        &self,
+        network: &crate::networks::Network,
+    ) -> eyre::Result<H256> {
+        let services = &self.services[network];
+        Self::fetch_latest_checkpoint_from_services(&services[..]).await
+    }
+
+    /// Fetch the latest checkpoint from a list of checkpoint fallback services.
+    pub async fn fetch_latest_checkpoint_from_services(
+        services: &[CheckpointFallbackService],
+    ) -> eyre::Result<H256> {
+        // Iterate over all mainnet checkpoint sync services and get the latest checkpoint slot for each.
+        let tasks: Vec<_> = services
+            .iter()
+            .map(|service| {
+                let service = service.clone();
+                tokio::spawn(async move {
+                    let client = reqwest::Client::new();
+                    let constructed_url = Self::construct_url(&service.endpoint);
+                    let res = client.get(&constructed_url).send().await?;
+                    let raw: RawSlotResponse = res.json().await?;
+                    if raw.data.slots.is_empty() {
+                        return Err(eyre::eyre!("no slots"));
+                    }
+                    Ok(raw.data.slots[0].clone())
+                })
+            })
+            .collect();
+        let slots = futures::future::join_all(tasks)
+            .await
+            .iter()
+            .filter_map(|slot| match &slot {
+                Ok(Ok(s)) => Some(s.clone()),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+
+        // Get the max epoch
+        let max_epoch_slot = slots.iter().max_by_key(|x| x.epoch).ok_or(eyre::eyre!(
+            "Failed to find max epoch from checkpoint slots"
+        ))?;
+        let max_epoch = max_epoch_slot.epoch;
+
+        // Filter out all the slots that are not the max epoch.
+        let slots = slots
+            .into_iter()
+            .filter(|x| x.epoch == max_epoch)
+            .collect::<Vec<_>>();
+
+        // Return the most commonly verified checkpoint.
+        let checkpoints = slots
+            .iter()
+            .filter_map(|x| x.block_root)
+            .collect::<Vec<_>>();
+        let mut m: HashMap<H256, usize> = HashMap::new();
+        for c in checkpoints {
+            *m.entry(c).or_default() += 1;
+        }
+        let most_common = m.into_iter().max_by_key(|(_, v)| *v).map(|(k, _)| k);
+
+        // Return the most commonly verified checkpoint for the latest epoch.
+        most_common.ok_or_else(|| eyre::eyre!("No checkpoint found"))
+    }
+
+    /// Associated function to fetch the latest checkpoint from a specific checkpoint sync fallback
+    /// service api url.
+    pub async fn fetch_checkpoint_from_api(url: &str) -> eyre::Result<H256> {
+        // Fetch the url
+        let client = reqwest::Client::new();
+        let constructed_url = Self::construct_url(url);
+        let res = client.get(constructed_url).send().await?;
+        let raw: RawSlotResponse = res.json().await?;
+        let slot = raw.data.slots[0].clone();
+        slot.block_root
+            .ok_or_else(|| eyre::eyre!("Checkpoint not in returned slot"))
+    }
+
+    /// Constructs the checkpoint fallback service url for fetching a slot.
+    ///
+    /// This is an associated function and can be used like so:
+    ///
+    /// ```rust
+    /// use config::CheckpointFallback;
+    ///
+    /// let url = CheckpointFallback::construct_url("https://sync-mainnet.beaconcha.in");
+    /// assert_eq!("https://sync-mainnet.beaconcha.in/checkpointz/v1/beacon/slots", url);
+    /// ```
+    pub fn construct_url(endpoint: &str) -> String {
+        format!("{}/checkpointz/v1/beacon/slots", endpoint)
+    }
+
+    /// Returns a list of all checkpoint fallback endpoints.
+    ///
+    /// ### Warning
+    ///
+    /// These services are not healthchecked **nor** trustworthy and may act with malice by returning invalid checkpoints.
+    pub fn get_all_fallback_endpoints(&self, network: &networks::Network) -> Vec<String> {
+        self.services[network]
+            .iter()
+            .map(|service| service.endpoint.clone())
+            .collect()
+    }
+
+    /// Returns a list of healthchecked checkpoint fallback endpoints.
+    ///
+    /// ### Warning
+    ///
+    /// These services are not trustworthy and may act with malice by returning invalid checkpoints.
+    pub fn get_healthy_fallback_endpoints(&self, network: &networks::Network) -> Vec<String> {
+        self.services[network]
+            .iter()
+            .filter(|service| service.state)
+            .map(|service| service.endpoint.clone())
+            .collect()
+    }
+}

config/src/cli.rs

@@ -0,0 +1,51 @@
+use std::{collections::HashMap, path::PathBuf};
+
+use figment::{providers::Serialized, value::Value};
+use serde::Serialize;
+
+/// Cli Config
+#[derive(Serialize)]
+pub struct CliConfig {
+    pub execution_rpc: Option<String>,
+    pub consensus_rpc: Option<String>,
+    pub checkpoint: Option<Vec<u8>>,
+    pub rpc_port: Option<u16>,
+    pub data_dir: PathBuf,
+    pub fallback: Option<String>,
+    pub load_external_fallback: bool,
+}
+
+impl CliConfig {
+    pub fn as_provider(&self, network: &str) -> Serialized<HashMap<&str, Value>> {
+        let mut user_dict = HashMap::new();
+
+        if let Some(rpc) = &self.execution_rpc {
+            user_dict.insert("execution_rpc", Value::from(rpc.clone()));
+        }
+
+        if let Some(rpc) = &self.consensus_rpc {
+            user_dict.insert("consensus_rpc", Value::from(rpc.clone()));
+        }
+
+        if let Some(checkpoint) = &self.checkpoint {
+            user_dict.insert("checkpoint", Value::from(hex::encode(checkpoint)));
+        }
+
+        if let Some(port) = self.rpc_port {
+            user_dict.insert("rpc_port", Value::from(port));
+        }
+
+        user_dict.insert("data_dir", Value::from(self.data_dir.to_str().unwrap()));
+
+        if let Some(fallback) = &self.fallback {
+            user_dict.insert("fallback", Value::from(fallback.clone()));
+        }
+
+        user_dict.insert(
+            "load_external_fallback",
+            Value::from(self.load_external_fallback),
+        );
+
+        Serialized::from(user_dict, network)
+    }
+}

config/src/config.rs

@@ -0,0 +1,100 @@
+use figment::{
+    providers::{Format, Serialized, Toml},
+    Figment,
+};
+use serde::{Deserialize, Serialize};
+use std::{path::PathBuf, process::exit};
+
+use crate::base::BaseConfig;
+use crate::cli::CliConfig;
+use crate::networks;
+use crate::types::{ChainConfig, Forks};
+use crate::utils::{bytes_deserialize, bytes_serialize};
+
+#[derive(Serialize, Deserialize, Debug, Default)]
+pub struct Config {
+    pub consensus_rpc: String,
+    pub execution_rpc: String,
+    pub rpc_port: Option<u16>,
+    #[serde(
+        deserialize_with = "bytes_deserialize",
+        serialize_with = "bytes_serialize"
+    )]
+    pub checkpoint: Vec<u8>,
+    pub data_dir: Option<PathBuf>,
+    pub chain: ChainConfig,
+    pub forks: Forks,
+    pub max_checkpoint_age: u64,
+    pub fallback: Option<String>,
+    pub load_external_fallback: bool,
+}
+
+impl Config {
+    pub fn from_file(config_path: &PathBuf, network: &str, cli_config: &CliConfig) -> Self {
+        let base_config = match network {
+            "mainnet" => networks::mainnet(),
+            "goerli" => networks::goerli(),
+            _ => BaseConfig::default(),
+        };
+
+        let base_provider = Serialized::from(base_config, network);
+        let toml_provider = Toml::file(config_path).nested();
+        let cli_provider = cli_config.as_provider(network);
+
+        let config_res = Figment::new()
+            .merge(base_provider)
+            .merge(toml_provider)
+            .merge(cli_provider)
+            .select(network)
+            .extract();
+
+        match config_res {
+            Ok(config) => config,
+            Err(err) => {
+                match err.kind {
+                    figment::error::Kind::MissingField(field) => {
+                        let field = field.replace('_', "-");
+
+                        println!(
+                            "\x1b[91merror\x1b[0m: missing configuration field: {}",
+                            field
+                        );
+
+                        println!(
+                            "\n\ttry supplying the propoper command line argument: --{}",
+                            field
+                        );
+
+                        println!("\talternatively, you can add the field to your helios.toml file or as an environment variable");
+                        println!("\nfor more information, check the github README");
+                    }
+                    _ => println!("cannot parse configuration: {}", err),
+                }
+                exit(1);
+            }
+        }
+    }
+
+    pub fn fork_version(&self, slot: u64) -> Vec<u8> {
+        let epoch = slot / 32;
+
+        if epoch >= self.forks.bellatrix.epoch {
+            self.forks.bellatrix.fork_version.clone()
+        } else if epoch >= self.forks.altair.epoch {
+            self.forks.altair.fork_version.clone()
+        } else {
+            self.forks.genesis.fork_version.clone()
+        }
+    }
+
+    pub fn to_base_config(&self) -> BaseConfig {
+        BaseConfig {
+            rpc_port: self.rpc_port.unwrap_or(8545),
+            consensus_rpc: Some(self.consensus_rpc.clone()),
+            checkpoint: self.checkpoint.clone(),
+            chain: self.chain.clone(),
+            forks: self.forks.clone(),
+            max_checkpoint_age: self.max_checkpoint_age,
+        }
+    }
+}

config/src/lib.rs

@@ -1,179 +1,26 @@
+/// Base Config
+pub mod base;
+pub use base::*;
+
+/// Core Config
+pub mod config;
+pub use crate::config::*;
+
+/// Checkpoint Config
+pub mod checkpoints;
+pub use checkpoints::*;
+
+/// Cli Config
+pub mod cli;
+pub use cli::*;
+
+/// Network Configuration
 pub mod networks;
+pub use networks::*;
 
-use std::{collections::HashMap, path::PathBuf, process::exit};
+/// Generic Config Types
+pub mod types;
+pub use types::*;
 
-use eyre::Result;
+/// Generic Utilities
-use figment::{
+pub mod utils;
-    providers::{Format, Serialized, Toml},
-    value::Value,
-    Figment,
-};
-use networks::BaseConfig;
-use serde::{Deserialize, Serialize};
-
-use common::utils::hex_str_to_bytes;
-
-#[derive(Serialize, Deserialize, Debug, Default)]
-pub struct Config {
-    pub consensus_rpc: String,
-    pub execution_rpc: String,
-    pub rpc_port: Option<u16>,
-    #[serde(
-        deserialize_with = "bytes_deserialize",
-        serialize_with = "bytes_serialize"
-    )]
-    pub checkpoint: Vec<u8>,
-    pub data_dir: Option<PathBuf>,
-    pub chain: ChainConfig,
-    pub forks: Forks,
-    pub max_checkpoint_age: u64,
-}
-
-impl Config {
-    pub fn from_file(config_path: &PathBuf, network: &str, cli_config: &CliConfig) -> Self {
-        let base_config = match network {
-            "mainnet" => networks::mainnet(),
-            "goerli" => networks::goerli(),
-            _ => BaseConfig::default(),
-        };
-
-        let base_provider = Serialized::from(base_config, network);
-        let toml_provider = Toml::file(config_path).nested();
-        let cli_provider = cli_config.as_provider(network);
-
-        let config_res = Figment::new()
-            .merge(base_provider)
-            .merge(toml_provider)
-            .merge(cli_provider)
-            .select(network)
-            .extract();
-
-        match config_res {
-            Ok(config) => config,
-            Err(err) => {
-                match err.kind {
-                    figment::error::Kind::MissingField(field) => {
-                        let field = field.replace('_', "-");
-
-                        println!(
-                            "\x1b[91merror\x1b[0m: missing configuration field: {}",
-                            field
-                        );
-
-                        println!(
-                            "\n\ttry supplying the propoper command line argument: --{}",
-                            field
-                        );
-
-                        println!("\talternatively, you can add the field to your helios.toml file or as an environment variable");
-                        println!("\nfor more information, check the github README");
-                    }
-                    _ => println!("cannot parse configuration: {}", err),
-                }
-                exit(1);
-            }
-        }
-    }
-
-    pub fn fork_version(&self, slot: u64) -> Vec<u8> {
-        let epoch = slot / 32;
-
-        if epoch >= self.forks.bellatrix.epoch {
-            self.forks.bellatrix.fork_version.clone()
-        } else if epoch >= self.forks.altair.epoch {
-            self.forks.altair.fork_version.clone()
-        } else {
-            self.forks.genesis.fork_version.clone()
-        }
-    }
-
-    pub fn to_base_config(&self) -> BaseConfig {
-        BaseConfig {
-            rpc_port: self.rpc_port.unwrap_or(8545),
-            consensus_rpc: Some(self.consensus_rpc.clone()),
-            checkpoint: self.checkpoint.clone(),
-            chain: self.chain.clone(),
-            forks: self.forks.clone(),
-            max_checkpoint_age: self.max_checkpoint_age,
-        }
-    }
-}
-
-#[derive(Serialize)]
-pub struct CliConfig {
-    pub execution_rpc: Option<String>,
-    pub consensus_rpc: Option<String>,
-    pub checkpoint: Option<Vec<u8>>,
-    pub rpc_port: Option<u16>,
-    pub data_dir: PathBuf,
-}
-
-impl CliConfig {
-    fn as_provider(&self, network: &str) -> Serialized<HashMap<&str, Value>> {
-        let mut user_dict = HashMap::new();
-
-        if let Some(rpc) = &self.execution_rpc {
-            user_dict.insert("execution_rpc", Value::from(rpc.clone()));
-        }
-
-        if let Some(rpc) = &self.consensus_rpc {
-            user_dict.insert("consensus_rpc", Value::from(rpc.clone()));
-        }
-
-        if let Some(checkpoint) = &self.checkpoint {
-            user_dict.insert("checkpoint", Value::from(hex::encode(checkpoint)));
-        }
-
-        if let Some(port) = self.rpc_port {
-            user_dict.insert("rpc_port", Value::from(port));
-        }
-
-        user_dict.insert("data_dir", Value::from(self.data_dir.to_str().unwrap()));
-
-        Serialized::from(user_dict, network)
-    }
-}
-
-#[derive(Serialize, Deserialize, Debug, Default, Clone)]
-pub struct ChainConfig {
-    pub chain_id: u64,
-    pub genesis_time: u64,
-    #[serde(
-        deserialize_with = "bytes_deserialize",
-        serialize_with = "bytes_serialize"
-    )]
-    pub genesis_root: Vec<u8>,
-}
-
-#[derive(Serialize, Deserialize, Debug, Default, Clone)]
-pub struct Forks {
-    pub genesis: Fork,
-    pub altair: Fork,
-    pub bellatrix: Fork,
-}
-
-#[derive(Serialize, Deserialize, Debug, Default, Clone)]
-pub struct Fork {
-    pub epoch: u64,
-    #[serde(
-        deserialize_with = "bytes_deserialize",
-        serialize_with = "bytes_serialize"
-    )]
-    pub fork_version: Vec<u8>,
-}
-
-fn bytes_deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
-where
-    D: serde::Deserializer<'de>,
-{
-    let bytes: String = serde::Deserialize::deserialize(deserializer)?;
-    Ok(hex_str_to_bytes(&bytes).unwrap())
-}
-
-fn bytes_serialize<S>(bytes: &Vec<u8>, serializer: S) -> Result<S::Ok, S::Error>
-where
-    S: serde::Serializer,
-{
-    let bytes_string = hex::encode(bytes);
-    serializer.serialize_str(&bytes_string)
-}

config/src/networks.rs

@@ -1,8 +1,24 @@
-use serde::Serialize;
-
-use crate::{bytes_serialize, ChainConfig, Fork, Forks};
 use common::utils::hex_str_to_bytes;
+use serde::{Deserialize, Serialize};
+use strum::{Display, EnumIter};
 
+use crate::base::BaseConfig;
+use crate::types::{ChainConfig, Fork, Forks};
+
+#[derive(
+    Debug,
+    Clone,
+    Copy,
+    Serialize,
+    Deserialize,
+    EnumIter,
+    Display,
+    Hash,
+    Eq,
+    PartialEq,
+    PartialOrd,
+    Ord,
+)]
 pub enum Network {
     MAINNET,
     GOERLI,
@@ -17,20 +33,6 @@ impl Network {
     }
 }
 
-#[derive(Serialize, Default)]
-pub struct BaseConfig {
-    pub rpc_port: u16,
-    pub consensus_rpc: Option<String>,
-    #[serde(
-        deserialize_with = "bytes_deserialize",
-        serialize_with = "bytes_serialize"
-    )]
-    pub checkpoint: Vec<u8>,
-    pub chain: ChainConfig,
-    pub forks: Forks,
-    pub max_checkpoint_age: u64,
-}
-
 pub fn mainnet() -> BaseConfig {
     BaseConfig {
         checkpoint: hex_str_to_bytes(

config/src/types.rs

@@ -0,0 +1,31 @@
+use serde::{Deserialize, Serialize};
+
+use crate::utils::{bytes_deserialize, bytes_serialize};
+
+#[derive(Serialize, Deserialize, Debug, Default, Clone)]
+pub struct ChainConfig {
+    pub chain_id: u64,
+    pub genesis_time: u64,
+    #[serde(
+        deserialize_with = "bytes_deserialize",
+        serialize_with = "bytes_serialize"
+    )]
+    pub genesis_root: Vec<u8>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default, Clone)]
+pub struct Forks {
+    pub genesis: Fork,
+    pub altair: Fork,
+    pub bellatrix: Fork,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default, Clone)]
+pub struct Fork {
+    pub epoch: u64,
+    #[serde(
+        deserialize_with = "bytes_deserialize",
+        serialize_with = "bytes_serialize"
+    )]
+    pub fork_version: Vec<u8>,
+}

config/src/utils.rs

@@ -0,0 +1,17 @@
+use common::utils::hex_str_to_bytes;
+
+pub fn bytes_deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let bytes: String = serde::Deserialize::deserialize(deserializer)?;
+    Ok(hex_str_to_bytes(&bytes).unwrap())
+}
+
+pub fn bytes_serialize<S>(bytes: &Vec<u8>, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    let bytes_string = hex::encode(bytes);
+    serializer.serialize_str(&bytes_string)
+}

config/tests/checkpoints.rs

@@ -0,0 +1,68 @@
+use config::networks;
+use ethers::types::H256;
+
+#[tokio::test]
+async fn test_checkpoint_fallback() {
+    let cf = config::checkpoints::CheckpointFallback::new();
+
+    assert_eq!(cf.services.get(&networks::Network::MAINNET), None);
+    assert_eq!(cf.services.get(&networks::Network::GOERLI), None);
+
+    assert_eq!(
+        cf.networks,
+        [networks::Network::MAINNET, networks::Network::GOERLI].to_vec()
+    );
+}
+
+#[tokio::test]
+async fn test_construct_checkpoints() {
+    let cf = config::checkpoints::CheckpointFallback::new()
+        .build()
+        .await
+        .unwrap();
+
+    assert!(cf.services[&networks::Network::MAINNET].len() > 1);
+    assert!(cf.services[&networks::Network::GOERLI].len() > 1);
+}
+
+#[tokio::test]
+async fn test_fetch_latest_checkpoints() {
+    let cf = config::checkpoints::CheckpointFallback::new()
+        .build()
+        .await
+        .unwrap();
+    let checkpoint = cf
+        .fetch_latest_checkpoint(&networks::Network::GOERLI)
+        .await
+        .unwrap();
+    assert!(checkpoint != H256::zero());
+    let checkpoint = cf
+        .fetch_latest_checkpoint(&networks::Network::MAINNET)
+        .await
+        .unwrap();
+    assert!(checkpoint != H256::zero());
+}
+
+#[tokio::test]
+async fn test_get_all_fallback_endpoints() {
+    let cf = config::checkpoints::CheckpointFallback::new()
+        .build()
+        .await
+        .unwrap();
+    let urls = cf.get_all_fallback_endpoints(&networks::Network::MAINNET);
+    assert!(urls.len() > 0);
+    let urls = cf.get_all_fallback_endpoints(&networks::Network::GOERLI);
+    assert!(urls.len() > 0);
+}
+
+#[tokio::test]
+async fn test_get_healthy_fallback_endpoints() {
+    let cf = config::checkpoints::CheckpointFallback::new()
+        .build()
+        .await
+        .unwrap();
+    let urls = cf.get_healthy_fallback_endpoints(&networks::Network::MAINNET);
+    assert!(urls.len() > 0);
+    let urls = cf.get_healthy_fallback_endpoints(&networks::Network::GOERLI);
+    assert!(urls.len() > 0);
+}

consensus/src/consensus.rs

@@ -94,6 +94,10 @@ impl<R: ConsensusRpc> ConsensusClient<R> {
     }
 
     pub async fn sync(&mut self) -> Result<()> {
+        info!(
+            "Consensus client in sync with checkpoint: 0x{}",
+            hex::encode(&self.initial_checkpoint)
+        );
         self.bootstrap().await?;
 
         let current_period = calc_sync_period(self.store.finalized_header.slot);