refactor: ease context sharing by using an inner arc pointer

Author: PJColombo

src/beacon_client/mod.rs

@@ -38,7 +38,6 @@ impl BeaconClient {
             Some(slot) => slot.to_string(),
             None => String::from("head"),
         };
-
         let url = self.build_url(&format!("eth/v2/beacon/blocks/{slot}"));
 
         let block_response = self.client.get(url).send().await?;

src/blobscan_client/types.rs

@@ -34,12 +34,13 @@ pub struct BlobEntity {
     pub index: u32,
 }
 
-#[derive(Serialize, Debug)]
+#[derive(Serialize, Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct FailedSlotsChunkEntity {
     pub initial_slot: u32,
     pub final_slot: u32,
 }
+
 #[derive(Serialize, Debug)]
 pub struct FailedSlotsChunksRequest {
     pub chunks: Vec<FailedSlotsChunkEntity>,

src/context.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use anyhow::Result;
 use ethers::prelude::*;
@@ -11,32 +11,53 @@ use crate::{
 use super::env::{get_env_vars, Environment};
 
 #[derive(Debug, Clone)]
-pub struct Context {
+struct ContextRef {
     pub beacon_client: BeaconClient,
     pub blobscan_client: BlobscanClient,
     pub provider: Provider<Http>,
 }
 
-pub fn create_context() -> Result<Context> {
-    let Environment {
-        blobscan_api_endpoint,
-        beacon_node_rpc,
-        execution_node_rpc,
-        secret_key,
-        ..
-    } = get_env_vars();
-    let request_timeout = Some(Duration::from_secs(8));
-
-    Ok(Context {
-        blobscan_client: BlobscanClient::try_from(BlobscanClientConfig {
-            base_url: blobscan_api_endpoint,
+#[derive(Debug, Clone)]
+pub struct Context {
+    inner: Arc<ContextRef>,
+}
+
+impl Context {
+    pub fn try_new() -> Result<Self> {
+        let Environment {
+            blobscan_api_endpoint,
+            beacon_node_rpc,
+            execution_node_rpc,
             secret_key,
-            timeout: request_timeout,
-        })?,
-        beacon_client: BeaconClient::try_from(BeaconClientConfig {
-            base_url: beacon_node_rpc,
-            timeout: request_timeout,
-        })?,
-        provider: Provider::<Http>::try_from(execution_node_rpc)?,
-    })
+            ..
+        } = get_env_vars();
+        let request_timeout = Some(Duration::from_secs(8));
+
+        Ok(Self {
+            inner: Arc::new(ContextRef {
+                blobscan_client: BlobscanClient::try_from(BlobscanClientConfig {
+                    base_url: blobscan_api_endpoint,
+                    secret_key,
+                    timeout: request_timeout,
+                })?,
+                beacon_client: BeaconClient::try_from(BeaconClientConfig {
+                    base_url: beacon_node_rpc,
+                    timeout: request_timeout,
+                })?,
+                provider: Provider::<Http>::try_from(execution_node_rpc)?,
+            }),
+        })
+    }
+
+    pub fn beacon_client(&self) -> &BeaconClient {
+        &self.inner.beacon_client
+    }
+
+    pub fn blobscan_client(&self) -> &BlobscanClient {
+        &self.inner.blobscan_client
+    }
+
+    pub fn provider(&self) -> &Provider<Http> {
+        &self.inner.provider
+    }
 }

src/main.rs

@@ -1,10 +1,8 @@
 use anyhow::Result;
+use context::Context;
 use slot_processor_manager::SlotProcessorManager;
 
-use crate::{
-    context::create_context,
-    utils::telemetry::{get_subscriber, init_subscriber},
-};
+use crate::utils::telemetry::{get_subscriber, init_subscriber};
 
 use std::{thread, time::Duration};
 
@@ -22,23 +20,25 @@ async fn main() -> Result<()> {
     let subscriber = get_subscriber("blobscan_indexer".into(), "info".into(), std::io::stdout);
     init_subscriber(subscriber);
 
-    let context = create_context()?;
-    let mut current_slot = match context.blobscan_client.get_slot().await? {
+    let context = Context::try_new()?;
+    let beacon_client = context.beacon_client();
+    let blobscan_client = context.blobscan_client();
+    let mut current_slot = match blobscan_client.get_slot().await? {
         Some(last_slot) => last_slot + 1,
         None => 0,
     };
     let slot_processor_manager = SlotProcessorManager::try_new(context.clone())?;
 
     loop {
-        if let Some(latest_beacon_block) = context.beacon_client.get_block(None).await? {
+        if let Some(latest_beacon_block) = beacon_client.get_block(None).await? {
             let latest_slot: u32 = latest_beacon_block.slot.parse()?;
 
             if current_slot < latest_slot {
                 slot_processor_manager
                     .process_slots(current_slot, latest_slot)
                     .await?;
 
-                context.blobscan_client.update_slot(latest_slot - 1).await?;
+                blobscan_client.update_slot(latest_slot - 1).await?;
 
                 current_slot = latest_slot;
             }

src/slot_processor_manager/mod.rs

@@ -1,5 +1,4 @@
 use core::panic;
-use std::sync::Arc;
 
 use futures::future::join_all;
 use tokio::task::{JoinError, JoinHandle};
@@ -11,17 +10,16 @@ use crate::{blobscan_client::types::FailedSlotsChunkEntity, context::Context};
 mod slot_processor;
 
 pub struct SlotProcessorManager {
-    shared_context: Arc<Context>,
+    context: Context,
     max_threads_length: u32,
 }
 
 impl SlotProcessorManager {
     pub fn try_new(context: Context) -> Result<Self, anyhow::Error> {
         let max_threads_length = std::thread::available_parallelism()?.get() as u32;
-        let shared_context = Arc::new(context);
 
         Ok(Self {
-            shared_context,
+            context,
             max_threads_length,
         })
     }
@@ -48,13 +46,13 @@ impl SlotProcessorManager {
                 slots_per_thread
             };
 
-            let thread_context = Arc::clone(&self.shared_context);
+            let thread_context = self.context.clone();
             let thread_initial_slot = current_slot;
             let thread_final_slot = current_slot + thread_slots_chunk;
 
             let thread = tokio::spawn(async move {
                 let slot_span = tracing::trace_span!("slot_processor", slot = end_slot);
-                let slot_processor = SlotProcessor::new(&thread_context);
+                let slot_processor = SlotProcessor::new(thread_context);
 
                 slot_processor
                     .process_slots(thread_initial_slot, thread_final_slot)
@@ -97,8 +95,8 @@ impl SlotProcessorManager {
             .collect::<Vec<FailedSlotsChunkEntity>>();
 
         if !failed_slots_chunks.is_empty() {
-            self.shared_context
-                .blobscan_client
+            self.context
+                .blobscan_client()
                 .add_failed_slots_chunks(failed_slots_chunks)
                 .await?;
         }

src/slot_processor_manager/slot_processor/mod.rs

@@ -2,8 +2,8 @@ use std::time::{Duration, Instant};
 
 use anyhow::{Context as AnyhowContext, Result};
 use backoff::{future::retry_notify, Error as BackoffError};
-use ethers::prelude::*;
 
+use ethers::prelude::*;
 use tracing::{error, info, warn};
 
 use crate::{
@@ -18,12 +18,12 @@ use self::helpers::{create_tx_hash_versioned_hashes_mapping, create_versioned_ha
 pub mod errors;
 mod helpers;
 
-pub struct SlotProcessor<'a> {
-    context: &'a Context,
+pub struct SlotProcessor {
+    context: Context,
 }
 
-impl<'a> SlotProcessor<'a> {
-    pub fn new(context: &'a Context) -> SlotProcessor {
+impl SlotProcessor {
+    pub fn new(context: Context) -> SlotProcessor {
         Self { context }
     }
 
@@ -71,11 +71,9 @@ impl<'a> SlotProcessor<'a> {
         &self,
         slot: u32,
     ) -> Result<(), backoff::Error<SingleSlotProcessingError>> {
-        let Context {
-            beacon_client,
-            blobscan_client,
-            provider,
-        } = self.context;
+        let beacon_client = self.context.beacon_client();
+        let blobscan_client = self.context.blobscan_client();
+        let provider = self.context.provider();
 
         let start = Instant::now();