refactor(sync): fetch blocks in batches (#189)

src/bin/dolos/common.rs

@@ -23,6 +23,7 @@ pub fn open_data_stores(config: &crate::Config) -> Result<Stores, Error> {
     let wal = wal::Store::open(
         rolldb_path.join("wal"),
         config.rolldb.k_param.unwrap_or(1000),
+        config.rolldb.immutable_overlap.clone(),
     )
     .map_err(Error::storage)?;
 

src/bin/dolos/main.rs

@@ -43,6 +43,7 @@ struct Cli {
 pub struct RolldbConfig {
     path: Option<std::path::PathBuf>,
     k_param: Option<u64>,
+    immutable_overlap: Option<u64>,
 }
 
 #[derive(Deserialize)]

src/submit/grpc/mod.rs

@@ -84,6 +84,7 @@ pub fn pipeline(
         let mut pull = sync::pull::Stage::new(
             config.peer_addresses[0].clone(),
             config.peer_magic,
+            50,
             sync::pull::Intersection::Tip,
         );
 

src/sync/ledger.rs

@@ -1,7 +1,6 @@
 use gasket::framework::*;
 use pallas::applying::{validate, Environment, UTxOs};
 use pallas::ledger::configs::{byron, shelley};
-use pallas::ledger::traverse::wellknown::GenesisValues;
 use pallas::ledger::traverse::{Era, MultiEraBlock, MultiEraInput, MultiEraOutput};
 use std::borrow::Cow;
 use std::collections::HashMap;

src/sync/mod.rs

@@ -20,6 +20,7 @@ pub mod roll;
 pub struct Config {
     pub peer_address: String,
     pub network_magic: u64,
+    pub block_fetch_batch_size: Option<usize>,
     pub network_id: u8,
     pub phase1_validation_enabled: bool,
 }
@@ -62,6 +63,7 @@ pub fn pipeline(
     let mut pull = pull::Stage::new(
         config.peer_address.clone(),
         config.network_magic,
+        config.block_fetch_batch_size.unwrap_or(50),
         pull_cursor,
     );
 
@@ -72,7 +74,6 @@ pub fn pipeline(
     info!(?cursor_ledger, "ledger cursor found");
 
     let mut roll = roll::Stage::new(wal, cursor_chain, cursor_ledger);
-
     let mut chain = chain::Stage::new(chain);
     let mut ledger = ledger::Stage::new(ledger, byron, shelley, config.phase1_validation_enabled);