feat: add max_slots_before_prune config

docs/pages/configuration/schema.mdx

@@ -65,17 +65,19 @@ The `upstream` section defines how to connect to the Ouroboros network to synchr
 
 The `storage` section controls how Dolos stores data in the local file system. This includes immutable chain blocks, the write ahead log and the ledger state.
 
-| property        | type    | example  |
-| --------------- | ------- | -------- |
-| path            | string  | "./data" |
-| wal_cache       | integer | 50       |
-| ledger_cache    | integer | 500      |
-| max_wal_history | integer | 10000    |
+| property               | type    | example  |
+| ---------------------- | ------- | -------- |
+| path                   | string  | "./data" |
+| wal_cache              | integer | 50       |
+| ledger_cache           | integer | 500      |
+| max_wal_history        | integer | 10000    |
+| max_slots_before_prune | integer | 10000    |
 
 - `path`: is the root directory where all data will be stored.
 - `wal_cache`: the size (in Mb) of the memory cache for the wal db.
 - `ledger_cache`: the size (in Mb) of the memory cache for the ledger db.
 - `max_wal_history`: the max number of slots to keep in the WAL.
+- `max_slots_before_prune`: the max number of slots to keep in the ledger store before pruning
 
 ## `genesis` section
 
@@ -91,7 +93,6 @@ Dolos requires Cardano genesis data to operate. For simplicity sake, we've decid
 - `shelley_path`: file path to the Shelley json genesis file
 - `alonzo_path`: file path to the Alonzo json genesis file
 
-
 ### `sync` section
 
 The `sync` section controls how Dolos synchronizes the chain from upstream peers. This involves fetch a batch of blocks from the upstream node and updating the corresponding local storage.
@@ -126,8 +127,8 @@ The `serve.grpc` section controls the options for the gRPC endpoint that can be
 
 The `serve.ouroboros` section controls the options for the Ouroboros mini-protocols endpoint that can be used by clients.
 
-| property       | type   | example           |
-| -------------- | ------ | ----------------- |
+| property    | type   | example        |
+| ----------- | ------ | -------------- |
 | listen_path | string | "dolos.socket" |
 
 - `listen_path`: the file path for the unix socket that will listen for Ouroboros node-to-client mini-protocols.
@@ -136,18 +137,18 @@ The `serve.ouroboros` section controls the options for the Ouroboros mini-protoc
 
 The `relay` section controls the options for handling inbound connection from other peers through Ouroboros node-to-node miniprotocols.
 
-| property       | type   | example           |
-| -------------- | ------ | ----------------- |
-| listen_address | string | "[::]:50051"      |
+| property       | type   | example      |
+| -------------- | ------ | ------------ |
+| listen_address | string | "[::]:50051" |
 
 - `listen_address`: the local address (`IP:PORT`) to listen for incoming Ouroboros connections (`[::]` represents any IP address).
 
 ## `logging` section
 
 The `logging` section controls the logging options to define the level of details to output.
 
-| property  | type   | example                                             |
-| --------- | ------ | --------------------------------------------------- |
-| max_level | option | "debug" / "info" / "warn" / "error"                 |
+| property       | type   | example                                        |
+| -------------- | ------ | ---------------------------------------------- |
+| max_level      | option | "debug" / "info" / "warn" / "error"            |
 | include_pallas | option | wheter to include logs from the Pallas library |
-| include_tonic | option | wheter to include logs from the Tonic library   |
+| include_tonic  | option | wheter to include logs from the Tonic library  |

src/bin/dolos/daemon.rs

@@ -23,6 +23,7 @@ pub async fn run(config: super::Config, _args: &Args) -> miette::Result<()> {
         mempool.clone(),
         &config.retries,
         false,
+        config.storage.max_slots_before_prune,
     )
     .into_diagnostic()
     .context("bootstrapping sync pipeline")?;

src/bin/dolos/doctor/rebuild_ledger.rs

@@ -81,9 +81,15 @@ pub fn run(config: &crate::Config, _args: &Args, feedback: &Feedback) -> miette:
             .into_diagnostic()
             .context("decoding blocks")?;
 
-        dolos::state::apply_block_batch(&blocks, &mut light, &byron, &shelley)
-            .into_diagnostic()
-            .context("importing blocks to ledger store")?;
+        dolos::state::apply_block_batch(
+            &blocks,
+            &mut light,
+            &byron,
+            &shelley,
+            config.storage.max_slots_before_prune,
+        )
+        .into_diagnostic()
+        .context("importing blocks to ledger store")?;
 
         blocks.last().inspect(|b| progress.set_position(b.slot()));
     }

src/bin/dolos/main.rs

@@ -75,6 +75,9 @@ pub struct StorageConfig {
 
     /// Maximum number of slots (not blocks) to keep in the WAL
     max_wal_history: Option<u64>,
+
+    /// Maximum number of slots to keep in the ledger before pruning
+    max_slots_before_prune: Option<u64>,
 }
 
 impl Default for StorageConfig {
@@ -84,6 +87,7 @@ impl Default for StorageConfig {
             wal_cache: None,
             ledger_cache: None,
             max_wal_history: None,
+            max_slots_before_prune: None,
         }
     }
 }

src/bin/dolos/sync.rs

@@ -26,6 +26,7 @@ pub fn run(config: &super::Config, args: &Args) -> miette::Result<()> {
         mempool,
         &config.retries,
         args.quit_on_tip,
+        config.storage.max_slots_before_prune,
     )
     .into_diagnostic()
     .context("bootstrapping sync pipeline")?;

src/state/mod.rs

@@ -220,6 +220,7 @@ pub fn apply_block_batch<'a>(
     store: &mut LedgerStore,
     byron: &byron::GenesisFile,
     shelley: &shelley::GenesisFile,
+    max_slots_before_prune: Option<u64>,
 ) -> Result<(), LedgerError> {
     let mut deltas: Vec<LedgerDelta> = vec![];
 
@@ -238,7 +239,10 @@ pub fn apply_block_batch<'a>(
         .map(|x| x.0)
         .unwrap();
 
-    let to_finalize = lastest_immutable_slot(tip, byron, shelley);
+    let to_finalize = max_slots_before_prune
+        .map(|x| tip - x)
+        .unwrap_or(lastest_immutable_slot(tip, byron, shelley));
+
     store.finalize(to_finalize)?;
 
     Ok(())

src/sync/apply.rs

@@ -17,6 +17,8 @@ pub struct Stage {
     shelley: shelley::GenesisFile,
     mempool: crate::mempool::Mempool, // Add this line
 
+    max_slots_before_prune: Option<u64>,
+
     pub upstream: UpstreamPort,
 
     #[metric]
@@ -33,6 +35,7 @@ impl Stage {
         mempool: crate::mempool::Mempool,
         byron: byron::GenesisFile,
         shelley: shelley::GenesisFile,
+        max_slots_before_prune: Option<u64>,
     ) -> Self {
         Self {
             wal,
@@ -43,6 +46,7 @@ impl Stage {
             upstream: Default::default(),
             block_count: Default::default(),
             wal_count: Default::default(),
+            max_slots_before_prune,
         }
     }
 
@@ -78,8 +82,14 @@ impl Stage {
 
         let block = MultiEraBlock::decode(body).or_panic()?;
 
-        crate::state::apply_block_batch([&block], &mut self.ledger, &self.byron, &self.shelley)
-            .or_panic()?;
+        crate::state::apply_block_batch(
+            [&block],
+            &mut self.ledger,
+            &self.byron,
+            &self.shelley,
+            self.max_slots_before_prune,
+        )
+        .or_panic()?;
 
         self.mempool.apply_block(&block);
 

src/sync/mod.rs

@@ -54,6 +54,7 @@ pub fn pipeline(
     mempool: Mempool,
     retries: &Option<gasket::retries::Policy>,
     quit_on_tip: bool,
+    max_slots_before_prune: Option<u64>,
 ) -> Result<Vec<gasket::runtime::Tether>, Error> {
     let mut pull = pull::Stage::new(
         upstream.peer_address.clone(),
@@ -65,7 +66,14 @@ pub fn pipeline(
 
     let mut roll = roll::Stage::new(wal.clone());
 
-    let mut apply = apply::Stage::new(wal.clone(), ledger, mempool.clone(), byron, shelley);
+    let mut apply = apply::Stage::new(
+        wal.clone(),
+        ledger,
+        mempool.clone(),
+        byron,
+        shelley,
+        max_slots_before_prune,
+    );
 
     let submit = submit::Stage::new(
         upstream.peer_address.clone(),