refactor(sync): fetch blocks in batches

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

@@ -29,7 +29,7 @@
     bytes: Vec<u8>,
 }
 
 impl Into<TxIdAndSize<EraTxId>> for Transaction {
     fn into(self) -> TxIdAndSize<EraTxId> {
         TxIdAndSize(
             EraTxId(self.era, self.hash.to_vec()),
@@ -38,7 +38,7 @@
     }
 }
 
 impl Into<EraTxBody> for Transaction {
     fn into(self) -> EraTxBody {
         EraTxBody(self.era, self.bytes)
     }
@@ -84,6 +84,7 @@
         let mut pull = sync::pull::Stage::new(
             config.peer_addresses[0].clone(),
             config.peer_magic,
+            50,
             sync::pull::Intersection::Tip,
         );
 

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);