fix: retry on Block parse errors when syncing

Signed-off-by: ljedrz <[email protected]>

snarkos/ledger/mod.rs

@@ -309,19 +309,21 @@ impl<N: Network> Ledger<N> {
                         let block_url = format!("{TARGET_URL}{height}.block");
 
                         // Fetch the bytes from the given url
-                        reqwest::get(block_url).await
+                        let block_bytes = reqwest::get(block_url).await?.bytes().await?;
+
+                        // Parse the block.
+                        let block = task::spawn_blocking(move || Block::from_bytes_le(&block_bytes)).await.unwrap()?;
+
+                        std::future::ready(Ok(block)).await
                     })
                 })
                 .buffered(CONCURRENT_REQUESTS)
-                .for_each(|response| async {
+                .for_each(|block| async {
+                    let block = block.unwrap();
                     // Use blocking tasks, as deserialization and adding blocks are expensive operations.
                     let self_clone = self.clone();
-                    let block_bytes = response.unwrap().bytes().await;
 
                     task::spawn_blocking(move || {
-                        // Parse the block.
-                        let block = Block::from_bytes_le(&block_bytes.unwrap()).unwrap();
-
                         // Add the block to the ledger.
                         self_clone.ledger.write().add_next_block(&block).unwrap();
 

snarkos/lib.rs

@@ -63,13 +63,14 @@ pub mod prelude {
     pub use snarkvm::prelude::{Address, Network};
 }
 
+use anyhow::anyhow;
 use backoff::{future::retry, ExponentialBackoff};
 use futures::Future;
 use std::time::Duration;
 
-pub(crate) async fn handle_dispatch_error<'a, T, F>(func: impl Fn() -> F + 'a) -> reqwest::Result<T>
+pub(crate) async fn handle_dispatch_error<'a, T, F>(func: impl Fn() -> F + 'a) -> anyhow::Result<T>
 where
-    F: Future<Output = Result<T, reqwest::Error>>,
+    F: Future<Output = Result<T, anyhow::Error>>,
 {
     fn default_backoff() -> ExponentialBackoff {
         ExponentialBackoff {
@@ -79,16 +80,26 @@ where
         }
     }
 
-    fn from_reqwest_err(err: reqwest::Error) -> backoff::Error<reqwest::Error> {
+    fn from_anyhow_err(err: anyhow::Error) -> backoff::Error<anyhow::Error> {
         use backoff::Error;
 
-        if err.is_timeout() {
-            debug!("Retrying server timeout error");
-            Error::Transient { err, retry_after: None }
+        if let Ok(err) = err.downcast::<reqwest::Error>() {
+            if err.is_timeout() {
+                debug!("Retrying server timeout error");
+                Error::Transient {
+                    err: err.into(),
+                    retry_after: None,
+                }
+            } else {
+                Error::Permanent(err.into())
+            }
         } else {
-            Error::Permanent(err)
+            Error::Transient {
+                err: anyhow!("Block parse error"),
+                retry_after: None,
+            }
         }
     }
 
-    retry(default_backoff(), || async { func().await.map_err(from_reqwest_err) }).await
+    retry(default_backoff(), || async { func().await.map_err(from_anyhow_err) }).await
 }