feat: add slot save interval for error resilience

Author: PJColombo

src/blobscan_client/mod.rs

@@ -61,7 +61,7 @@ impl BlobscanClient {
 
         let index_response = self
             .client
-            .post(url)
+            .put(url)
             .bearer_auth(token)
             .json(&index_request)
             .send()

src/main.rs

@@ -15,6 +15,8 @@ mod env;
 mod slots_processor;
 mod utils;
 
+const MAX_SLOTS_PER_SAVE: u32 = 1000;
+
 #[tokio::main]
 async fn main() -> Result<()> {
     dotenv::dotenv().ok();
@@ -42,21 +44,60 @@ async fn main() -> Result<()> {
             let latest_slot: u32 = latest_beacon_block.slot.parse()?;
 
             if current_slot < latest_slot {
-                let slot_manager_span = tracing::debug_span!(
-                    "slot_processor_manager",
-                    initial_slot = current_slot,
-                    final_slot = latest_slot
+                let unprocessed_slots = latest_slot - current_slot;
+                let current_max_slots_size = std::cmp::min(unprocessed_slots, MAX_SLOTS_PER_SAVE);
+                let num_chunks = unprocessed_slots / current_max_slots_size;
+
+                let remaining_slots = unprocessed_slots % current_max_slots_size;
+                let num_chunks = if remaining_slots > 0 {
+                    num_chunks + 1
+                } else {
+                    num_chunks
+                };
+
+                info!(
+                    "Processing slots from {} to {}, partitioned into {} chunks…",
+                    current_slot, latest_slot, num_chunks
                 );
 
-                slots_processor
-                    .process_slots(current_slot, latest_slot)
-                    .instrument(slot_manager_span)
-                    .await?;
+                for i in 0..num_chunks {
+                    let slots_in_current_chunk = if i == num_chunks - 1 {
+                        current_max_slots_size + remaining_slots
+                    } else {
+                        current_max_slots_size
+                    };
+
+                    let chunk_initial_slot = current_slot + i * current_max_slots_size;
+                    let chunk_final_slot = chunk_initial_slot + slots_in_current_chunk;
+
+                    let slot_manager_span = tracing::info_span!(
+                        "slots_processor",
+                        initial_slot = chunk_initial_slot,
+                        final_slot = chunk_final_slot
+                    );
 
-                blobscan_client.update_slot(latest_slot - 1).await?;
-                info!("Latest slot updated to {}", latest_slot - 1);
+                    slots_processor
+                        .process_slots(chunk_initial_slot, chunk_final_slot)
+                        .instrument(slot_manager_span)
+                        .await?;
+
+                    blobscan_client.update_slot(chunk_final_slot - 1).await?;
+
+                    info!(
+                        "Chunk {} of {} ({} slots) processed successfully!. Updating latest slot to {}.",
+                        i+1,
+                        num_chunks,
+                        chunk_final_slot - chunk_initial_slot,
+                        chunk_final_slot - 1
+                    );
+                }
 
                 current_slot = latest_slot;
+
+                info!(
+                    "All slots processed successfully! Total slots processed: {}",
+                    unprocessed_slots
+                );
             }
         }
 

src/slots_processor/error.rs

@@ -1,4 +1,4 @@
-use super::slot_processor::errors::SlotProcessorError;
+use super::slot_processor::error::SlotProcessorError;
 
 #[derive(Debug, thiserror::Error)]
 pub enum SlotsChunkThreadError {

src/slots_processor/mod.rs

@@ -45,28 +45,30 @@ impl SlotsProcessor {
             return Ok(());
         }
 
-        let slots_chunk = end_slot - start_slot;
-        let slots_per_thread = slots_chunk / self.config.threads_length;
-        let threads_length = if slots_per_thread > 0 {
-            self.config.threads_length
+        let unprocessed_slots = end_slot - start_slot;
+        let num_threads = std::cmp::min(self.config.threads_length, unprocessed_slots);
+        let slots_per_thread = unprocessed_slots / num_threads;
+        let remaining_slots = unprocessed_slots % num_threads;
+        let num_threads = if slots_per_thread > 0 {
+            num_threads
         } else {
-            slots_chunk
+            unprocessed_slots
         };
+
         let mut handles: Vec<JoinHandle<Result<(), SlotsChunkThreadError>>> = vec![];
-        let mut current_slot = start_slot;
 
-        for i in 0..threads_length {
-            let thread_slots_chunk = if i == 0 {
-                slots_per_thread + slots_chunk % self.config.threads_length
+        for i in 0..num_threads {
+            let slots_in_current_thread = if i == num_threads - 1 {
+                slots_per_thread + remaining_slots
             } else {
                 slots_per_thread
             };
 
             let thread_context = self.context.clone();
-            let thread_initial_slot = current_slot;
-            let thread_final_slot = current_slot + thread_slots_chunk;
+            let thread_initial_slot = start_slot + i * slots_per_thread;
+            let thread_final_slot = thread_initial_slot + slots_in_current_thread;
 
-            let thread_slots_span = tracing::info_span!(
+            let thread_slots_span = tracing::debug_span!(
                 "slots_chunk_processor",
                 chunk_initial_slot = thread_initial_slot,
                 chunk_final_slot = thread_final_slot
@@ -102,8 +104,6 @@ impl SlotsProcessor {
             );
 
             handles.push(handle);
-
-            current_slot += thread_slots_chunk;
         }
 
         let handle_outputs = join_all(handles).await;

src/slots_processor/slot_processor/errors.rs renamed to src/slots_processor/slot_processor/error.rs

@@ -12,10 +12,10 @@ use crate::{
     utils::exp_backoff::get_exp_backoff_config,
 };
 
-use self::errors::SlotProcessorError;
+use self::error::SlotProcessorError;
 use self::helpers::{create_tx_hash_versioned_hashes_mapping, create_versioned_hash_blob_mapping};
 
-pub mod errors;
+pub mod error;
 mod helpers;
 
 pub struct SlotProcessor {