refactor: Resolves #42 replace anyhow with LakeError enum (thiserror)

lake-framework/Cargo.toml

@@ -4,7 +4,6 @@ version = "0.0.0" # managed by cargo-workspaces
 edition = "2021"
 
 [dependencies]
-anyhow = "1.0.51"
 aws-config = "0.53.0"
 aws-types = "0.53.0"
 aws-credential-types = "0.53.0"
@@ -24,6 +23,8 @@ near-lake-primitives = { path = "../lake-primitives" }
 
 [dev-dependencies]
 aws-smithy-http = "0.53.0"
+# use by examples
+anyhow = "1.0.51"
 
 # used by nft_indexer example
 regex = "1.5.4"

lake-framework/README.md

@@ -12,7 +12,8 @@ fn main() -> anyhow::Result<()> {
         .testnet()
         .start_block_height(112205773)
         .build()?
-        .run(handle_block)
+        .run(handle_block)?;
+    Ok(())
 }
 
 // The handler function to take the `Block`
@@ -45,7 +46,9 @@ fn main() -> anyhow::Result<()> {
         .testnet()
         .start_block_height(112205773)
         .build()?
-        .run_with_context(handle_block, &context)
+        .run_with_context(handle_block, &context)?;
+
+    Ok(())
 }
 
 // The handler function to take the `Block`

lake-framework/examples/actions.rs

@@ -14,7 +14,9 @@ fn main() -> anyhow::Result<()> {
         .mainnet()
         .start_block_height(88444526)
         .build()?
-        .run(print_function_calls_to_my_account) // developer-defined async function that handles each block
+        // developer-defined async function that handles each block
+        .run(print_function_calls_to_my_account)?;
+    Ok(())
 }
 
 async fn print_function_calls_to_my_account(

lake-framework/examples/nft_indexer.rs

@@ -20,7 +20,8 @@ fn main() -> anyhow::Result<()> {
         .testnet()
         .start_block_height(112205773)
         .build()?
-        .run(handle_block) // developer-defined async function that handles each block
+        .run(handle_block)?; // developer-defined async function that handles each block
+    Ok(())
 }
 
 async fn handle_block(mut block: near_lake_primitives::block::Block) -> anyhow::Result<()> {

lake-framework/examples/simple.rs

@@ -10,7 +10,8 @@ fn main() -> anyhow::Result<()> {
         .testnet()
         .start_block_height(112205773)
         .build()?
-        .run(handle_block) // developer-defined async function that handles each block
+        .run(handle_block)?; // developer-defined async function that handles each block
+    Ok(())
 }
 
 async fn handle_block(block: near_lake_primitives::block::Block) -> anyhow::Result<()> {

lake-framework/examples/with_context.rs

@@ -40,7 +40,8 @@ fn main() -> anyhow::Result<()> {
         .start_block_height(88444526)
         .build()?
         // developer-defined async function that handles each block
-        .run_with_context(print_function_calls_to_my_account, &context)
+        .run_with_context(print_function_calls_to_my_account, &context)?;
+    Ok(())
 }
 
 async fn print_function_calls_to_my_account(

lake-framework/src/lib.rs

@@ -7,7 +7,7 @@ use futures::{Future, StreamExt};
 pub use near_lake_primitives::{self, near_indexer_primitives, LakeContext};
 
 pub use aws_credential_types::Credentials;
-pub use types::{Lake, LakeBuilder};
+pub use types::{Lake, LakeBuilder, LakeError};
 
 mod s3_fetchers;
 mod streamer;
@@ -31,20 +31,23 @@ impl types::Lake {
     ///        .testnet()
     ///        .start_block_height(112205773)
     ///        .build()?
-    ///        .run_with_context(handle_block, &context)
+    ///        .run_with_context(handle_block, &context)?;
+    ///    Ok(())
     ///# }
     ///
     /// # async fn handle_block(_block: near_lake_primitives::block::Block, context: &MyContext) -> anyhow::Result<()> { Ok(()) }
     ///```
-    pub fn run_with_context<'context, C, Fut>(
+    pub fn run_with_context<'context, C, E, Fut>(
         self,
         f: impl Fn(near_lake_primitives::block::Block, &'context C) -> Fut,
         context: &'context C,
-    ) -> anyhow::Result<()>
+    ) -> Result<(), LakeError>
     where
-        Fut: Future<Output = anyhow::Result<()>>,
+        Fut: Future<Output = Result<(), E>>,
+        E: Into<Box<dyn std::error::Error>>,
     {
-        let runtime = tokio::runtime::Runtime::new()?;
+        let runtime = tokio::runtime::Runtime::new()
+            .map_err(|err| LakeError::RuntimeStartError { error: err })?;
 
         runtime.block_on(async move {
             // instantiate the NEAR Lake Framework Stream
@@ -65,8 +68,8 @@ impl types::Lake {
             // propagate errors from the sender
             match sender.await {
                 Ok(Ok(())) => Ok(()),
-                Ok(Err(e)) => Err(e),
-                Err(e) => Err(anyhow::Error::from(e)), // JoinError
+                Ok(Err(err)) => Err(err),
+                Err(err) => Err(err.into()), // JoinError
             }
         })
     }
@@ -78,17 +81,19 @@ impl types::Lake {
     ///        .testnet()
     ///        .start_block_height(112205773)
     ///        .build()?
-    ///        .run(handle_block)
+    ///        .run(handle_block)?;
+    ///    Ok(())
     ///# }
     ///
     /// # async fn handle_block(_block: near_lake_primitives::block::Block) -> anyhow::Result<()> { Ok(()) }
     ///```
-    pub fn run<Fut>(
+    pub fn run<Fut, E>(
         self,
         f: impl Fn(near_lake_primitives::block::Block) -> Fut,
-    ) -> anyhow::Result<()>
+    ) -> Result<(), LakeError>
     where
-        Fut: Future<Output = anyhow::Result<()>>,
+        Fut: Future<Output = Result<(), E>>,
+        E: Into<Box<dyn std::error::Error>>,
     {
         self.run_with_context(|block, _context| f(block), &())
     }

lake-framework/src/s3_fetchers.rs

@@ -77,10 +77,7 @@ pub(crate) async fn list_block_heights(
     lake_s3_client: &impl S3Client,
     s3_bucket_name: &str,
     start_from_block_height: crate::types::BlockHeight,
-) -> Result<
-    Vec<crate::types::BlockHeight>,
-    crate::types::LakeError<aws_sdk_s3::error::ListObjectsV2Error>,
-> {
+) -> Result<Vec<crate::types::BlockHeight>, crate::types::LakeError> {
     tracing::debug!(
         target: crate::LAKE_FRAMEWORK,
         "Fetching block heights from S3, after #{}...",
@@ -117,10 +114,7 @@ pub(crate) async fn fetch_streamer_message(
     lake_s3_client: &impl S3Client,
     s3_bucket_name: &str,
     block_height: crate::types::BlockHeight,
-) -> Result<
-    near_lake_primitives::StreamerMessage,
-    crate::types::LakeError<aws_sdk_s3::error::GetObjectError>,
-> {
+) -> Result<near_lake_primitives::StreamerMessage, crate::types::LakeError> {
     let block_view = {
         let body_bytes = loop {
             match lake_s3_client
@@ -177,10 +171,7 @@ async fn fetch_shard_or_retry(
     s3_bucket_name: &str,
     block_height: crate::types::BlockHeight,
     shard_id: u64,
-) -> Result<
-    near_lake_primitives::IndexerShard,
-    crate::types::LakeError<aws_sdk_s3::error::GetObjectError>,
-> {
+) -> Result<near_lake_primitives::IndexerShard, crate::types::LakeError> {
     let body_bytes = loop {
         match lake_s3_client
             .get_object(

lake-framework/src/streamer.rs

@@ -15,7 +15,7 @@ use crate::{s3_fetchers, types};
 pub(crate) fn streamer(
     config: crate::Lake,
 ) -> (
-    tokio::task::JoinHandle<Result<(), anyhow::Error>>,
+    tokio::task::JoinHandle<Result<(), crate::types::LakeError>>,
     mpsc::Receiver<near_indexer_primitives::StreamerMessage>,
 ) {
     let (sender, receiver) = mpsc::channel(config.blocks_preload_pool_size);
@@ -80,7 +80,7 @@ async fn prefetch_block_heights_into_pool(
     >,
     limit: usize,
     await_for_at_least_one: bool,
-) -> anyhow::Result<Vec<crate::types::BlockHeight>> {
+) -> Result<Vec<crate::types::BlockHeight>, crate::types::LakeError> {
     let mut block_heights = Vec::with_capacity(limit);
     for remaining_limit in (0..limit).rev() {
         tracing::debug!(target: crate::LAKE_FRAMEWORK, "Polling for the next block height without awaiting... (up to {} block heights are going to be fetched)", remaining_limit);
@@ -96,7 +96,9 @@ async fn prefetch_block_heights_into_pool(
                             block_heights.push(block_height);
                         }
                         None => {
-                            return Err(anyhow::anyhow!("This state should be unreachable as the block heights stream should be infinite."));
+                            return Err(crate::types::LakeError::InternalError {
+                                error_message: "This state should be unreachable as the block heights stream should be infinite.".to_string()
+                            });
                         }
                     }
                     continue;
@@ -105,7 +107,11 @@ async fn prefetch_block_heights_into_pool(
                 break;
             }
             std::task::Poll::Ready(None) => {
-                return Err(anyhow::anyhow!("This state should be unreachable as the block heights stream should be infinite."));
+                return Err(
+                    crate::types::LakeError::InternalError {
+                        error_message: "This state should be unreachable as the block heights stream should be infinite.".to_string()
+                    }
+                );
             }
         }
     }
@@ -116,7 +122,7 @@ async fn prefetch_block_heights_into_pool(
 pub(crate) async fn start(
     streamer_message_sink: mpsc::Sender<near_indexer_primitives::StreamerMessage>,
     config: crate::Lake,
-) -> anyhow::Result<()> {
+) -> Result<(), crate::types::LakeError> {
     let mut start_from_block_height = config.start_block_height;
 
     let s3_client = if let Some(config) = config.s3_config {
@@ -235,7 +241,7 @@ pub(crate) async fn start(
             let streamer_message_sink_send_future = streamer_message_sink.send(streamer_message);
 
             let (prefetch_res, send_res): (
-                Result<Vec<types::BlockHeight>, anyhow::Error>,
+                Result<Vec<types::BlockHeight>, crate::types::LakeError>,
                 Result<_, SendError<near_indexer_primitives::StreamerMessage>>,
             ) = futures::join!(
                 prefetched_block_heights_future,

lake-framework/src/types.rs

@@ -112,20 +112,37 @@ impl LakeBuilder {
 
 #[allow(clippy::enum_variant_names)]
 #[derive(thiserror::Error, Debug)]
-pub enum LakeError<E> {
+pub enum LakeError {
     #[error("Failed to parse structure from JSON: {error_message}")]
     ParseError {
         #[from]
         error_message: serde_json::Error,
     },
     #[error("AWS S3 error")]
-    AwsError {
+    AwsGetObjectError {
         #[from]
-        error: aws_sdk_s3::types::SdkError<E>,
+        error: aws_sdk_s3::types::SdkError<aws_sdk_s3::error::GetObjectError>,
+    },
+    #[error("AWS S3 error")]
+    AwsLisObjectsV2Error {
+        #[from]
+        error: aws_sdk_s3::types::SdkError<aws_sdk_s3::error::ListObjectsV2Error>,
     },
     #[error("Failed to convert integer")]
     IntConversionError {
         #[from]
         error: std::num::TryFromIntError,
     },
+    #[error("Join error")]
+    JoinError {
+        #[from]
+        error: tokio::task::JoinError,
+    },
+    #[error("Failed to start runtime")]
+    RuntimeStartError {
+        #[from]
+        error: std::io::Error,
+    },
+    #[error("Internal error: {error_message}")]
+    InternalError { error_message: String },
 }