wip

Author: kariy

crates/node/src/full/mod.rs

@@ -121,7 +121,7 @@ impl Node {
 
         // --- build pipeline
 
-        let (mut pipeline, pipeline_handle) = Pipeline::new(storage_provider.clone(), 50);
+        let (mut pipeline, pipeline_handle) = Pipeline::new(storage_provider.clone(), 256);
         let block_downloader = BatchBlockDownloader::new_gateway(gateway_client.clone(), 20);
         pipeline.add_stage(Blocks::new(storage_provider.clone(), block_downloader));
         pipeline.add_stage(Classes::new(storage_provider.clone(), gateway_client.clone(), 20));

crates/sync/pipeline/src/lib.rs

@@ -404,7 +404,7 @@ where
 
             let provider_mut = self.storage_provider.provider_mut();
             provider_mut.set_checkpoint(id, last_block_processed)?;
-            provider_mut.commit();
+            provider_mut.commit()?;
 
             last_block_processed_list.push(last_block_processed);
             info!(target: "pipeline", checkpoint = %last_block_processed, "New checkpoint set.");

crates/sync/stage/src/downloader.rs

@@ -185,15 +185,7 @@ where
     }
 
     async fn download_batch(&self, keys: &[D::Key]) -> Vec<DownloaderResult<D::Value, D::Error>> {
-        join_all(keys.iter().map(|key| async move {
-            let start = std::time::Instant::now();
-            let result = self.downloader.download(key).await;
-            let elapsed = start.elapsed();
-            println!("elapsed_ms = {}", elapsed.as_millis());
-
-            result
-        }))
-        .await
+        join_all(keys.iter().map(|key| self.downloader.download(key))).await
     }
 
     async fn download_batch_with_retry(
@@ -234,7 +226,7 @@ where
             }
 
             // if not failed keys, all requests succeeded
-            if dbg!(failed_keys.is_empty()) {
+            if failed_keys.is_empty() {
                 break;
             }