feat(phoenix): get payload from all

src/beacon_api.rs

@@ -2,6 +2,7 @@ use anyhow::anyhow;
 use rand::seq::SliceRandom;
 use reqwest::{StatusCode, Url};
 use serde::Deserialize;
+use tracing::warn;
 
 #[derive(Deserialize)]
 struct BeaconResponse<T> {
@@ -72,7 +73,7 @@ impl BeaconApi {
         self.nodes.choose(&mut rand::thread_rng()).unwrap()
     }
 
-    pub async fn get_validator_index(&self, pubkey: &String) -> reqwest::Result<String> {
+    pub async fn validator_index(&self, pubkey: &String) -> reqwest::Result<String> {
         let url = format!(
             "{}eth/v1/beacon/states/head/validators/{}",
             self.get_node(),
@@ -87,11 +88,14 @@ impl BeaconApi {
             .map(|body| body.data.index)
     }
 
-    pub async fn get_payload(&self, slot: &i64) -> anyhow::Result<Option<ExecutionPayload>> {
-        let url = format!("{}eth/v2/beacon/blocks/{}", self.get_node(), slot);
-
-        let res = self.client.get(url).send().await?;
-
+    // Method to fetch the payload from a node and a slot
+    async fn fetch_payload(
+        &self,
+        node: &Url,
+        slot: i64,
+    ) -> anyhow::Result<Option<ExecutionPayload>> {
+        let url = format!("{}eth/v2/beacon/blocks/{}", node, slot);
+        let res = self.client.get(&url).send().await?;
         match res.status() {
             StatusCode::NOT_FOUND => Ok(None),
             StatusCode::OK => {
@@ -102,23 +106,41 @@ impl BeaconApi {
                 Ok(Some(block.body.execution_payload))
             }
             status => Err(anyhow!(
-                "failed to fetch block_hash by slot. slot = {} status = {} url = {}",
+                "failed to fetch block by slot. slot = {} status = {} url = {}",
                 slot,
                 status,
                 res.url()
             )),
         }
     }
 
-    pub async fn get_sync_status(&self, node_url: &Url) -> reqwest::Result<SyncStatus> {
+    // Method to fetch the sync status from a node
+    pub async fn fetch_sync_status(&self, node_url: &Url) -> reqwest::Result<SyncStatus> {
         let url = format!("{}eth/v1/node/syncing", node_url);
         self.client
-            .get(url)
+            .get(&url)
             .send()
             .await?
             .error_for_status()?
             .json::<BeaconResponse<SyncStatus>>()
             .await
             .map(|body| body.data)
+            .map_err(Into::into)
+    }
+
+    pub async fn fetch_payload_all(&self, slot: i64) -> anyhow::Result<Option<ExecutionPayload>> {
+        for (i, node) in self.nodes.iter().enumerate() {
+            match self.fetch_payload(node, slot).await {
+                Ok(res) => return Ok(res),
+                Err(err) => {
+                    warn!("failed to fetch payload from {}: {:?}", node, err);
+                    if i == self.nodes.len() - 1 {
+                        return Err(err);
+                    }
+                }
+            }
+        }
+
+        unreachable!("last iteration should always return")
     }
 }

src/phoenix/consensus_node.rs

@@ -21,7 +21,7 @@ impl ConsensusNodeMonitor {
         let mut results = Vec::new();
 
         for url in &APP_CONFIG.consensus_nodes {
-            let status = self.beacon_api.get_sync_status(url).await;
+            let status = self.beacon_api.fetch_sync_status(url).await;
 
             match status {
                 Ok(s) => results.push(!s.is_syncing),

src/phoenix/inclusion_monitor/mod.rs

@@ -324,7 +324,7 @@ async fn was_attempted_reorg(
     delivered: &DeliveredPayload,
 ) -> anyhow::Result<bool> {
     let prev_slot = delivered.slot - 1;
-    let prev_payload = beacon_api.get_payload(&prev_slot).await?;
+    let prev_payload = beacon_api.fetch_payload_all(prev_slot).await?;
     Ok(prev_payload
         .map(|p| p.block_number == delivered.block_number)
         .unwrap_or(false))
@@ -337,7 +337,7 @@ async fn check_missing_payload(
     payload: &DeliveredPayload,
     relay_pool: &PgPool,
 ) -> anyhow::Result<()> {
-    let block = beacon_api.get_payload(&payload.slot).await?;
+    let block = beacon_api.fetch_payload_all(payload.slot).await?;
 
     match block {
         Some(ExecutionPayload { block_hash, .. }) => {

src/serve/validator.rs

@@ -159,7 +159,7 @@ pub async fn validator_registrations(
 
     for v in non_cached {
         futs.push(async move {
-            let idx = beacon_api.get_validator_index(&v.pubkey).await.unwrap();
+            let idx = beacon_api.validator_index(&v.pubkey).await.unwrap();
 
             ValidatorWithIndex {
                 inserted_at: v.inserted_at,