feat: allow configurable ledger prune height (#440)

Co-authored-by: Martin Schere <tinschere@gmail.com>

docs/pages/configuration/schema.mdx

@@ -66,17 +66,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_ledger_history     | 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_ledger_history`: the max number of slots to keep in the ledger store before pruning. If not set, the ledger will prune past the immutable slot.
 
 ## `genesis` section
 
@@ -130,8 +132,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.
@@ -140,18 +142,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

@@ -18,6 +18,7 @@ pub async fn run(config: super::Config, _args: &Args) -> miette::Result<()> {
     let sync = dolos::sync::pipeline(
         &config.sync,
         &config.upstream,
+        &config.storage,
         wal.clone(),
         ledger.clone(),
         genesis.clone(),

src/bin/dolos/doctor/rebuild_ledger.rs

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

src/bin/dolos/main.rs

@@ -63,32 +63,6 @@ struct Cli {
     config: Option<std::path::PathBuf>,
 }
 
-#[derive(Serialize, Deserialize)]
-pub struct StorageConfig {
-    path: std::path::PathBuf,
-
-    /// Size (in Mb) of memory allocated for WAL caching
-    wal_cache: Option<usize>,
-
-    /// Size (in Mb) of memory allocated for ledger caching
-    ledger_cache: Option<usize>,
-
-    /// Maximum number of slots (not blocks) to keep in the WAL
-    max_wal_history: Option<u64>,
-}
-
-impl Default for StorageConfig {
-    fn default() -> Self {
-        Self {
-            path: PathBuf::from("data"),
-            wal_cache: None,
-            ledger_cache: None,
-            max_wal_history: None,
-        }
-    }
-}
-
-// TODO: add hash of genesis for runtime verification
 #[derive(Serialize, Deserialize)]
 pub struct GenesisConfig {
     byron_path: PathBuf,
@@ -151,7 +125,7 @@ impl Default for LoggingConfig {
 #[derive(Serialize, Deserialize)]
 pub struct Config {
     pub upstream: dolos::model::UpstreamConfig,
-    pub storage: StorageConfig,
+    pub storage: dolos::model::StorageConfig,
     pub genesis: GenesisConfig,
     pub sync: dolos::sync::Config,
     pub submit: dolos::model::SubmitConfig,

src/bin/dolos/sync.rs

@@ -19,6 +19,7 @@ pub fn run(config: &super::Config, args: &Args) -> miette::Result<()> {
     let sync = dolos::sync::pipeline(
         &config.sync,
         &config.upstream,
+        &config.storage,
         wal,
         ledger,
         genesis,

src/model.rs

@@ -42,3 +42,32 @@ pub struct UpstreamConfig {
 pub struct SubmitConfig {
     pub prune_height: Option<u64>,
 }
+
+#[derive(Serialize, Deserialize)]
+pub struct StorageConfig {
+    pub path: std::path::PathBuf,
+
+    /// Size (in Mb) of memory allocated for WAL caching
+    pub wal_cache: Option<usize>,
+
+    /// Size (in Mb) of memory allocated for ledger caching
+    pub ledger_cache: Option<usize>,
+
+    /// Maximum number of slots (not blocks) to keep in the WAL
+    pub max_wal_history: Option<u64>,
+
+    /// Maximum number of slots to keep in the ledger before pruning
+    pub max_ledger_history: Option<u64>,
+}
+
+impl Default for StorageConfig {
+    fn default() -> Self {
+        Self {
+            path: std::path::PathBuf::from("data"),
+            wal_cache: None,
+            ledger_cache: None,
+            max_wal_history: None,
+            max_ledger_history: None,
+        }
+    }
+}

src/state/mod.rs

@@ -220,6 +220,7 @@ pub fn apply_block_batch<'a>(
     blocks: impl IntoIterator<Item = &'a MultiEraBlock<'a>>,
     store: &LedgerStore,
     genesis: &Genesis,
+    max_ledger_history: 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, genesis);
+    let to_finalize = max_ledger_history
+        .map(|x| tip - x)
+        .unwrap_or(lastest_immutable_slot(tip, genesis));
+
     store.finalize(to_finalize)?;
 
     Ok(())

src/sync/apply.rs

@@ -18,6 +18,8 @@ pub struct Stage {
     genesis: Arc<Genesis>,
     mempool: crate::mempool::Mempool, // Add this line
 
+    max_ledger_history: Option<u64>,
+
     pub upstream: UpstreamPort,
 
     #[metric]
@@ -33,12 +35,14 @@ impl Stage {
         ledger: crate::state::LedgerStore,
         mempool: crate::mempool::Mempool,
         genesis: Arc<Genesis>,
+        max_ledger_history: Option<u64>,
     ) -> Self {
         Self {
             wal,
             ledger,
             mempool,
             genesis,
+            max_ledger_history,
             upstream: Default::default(),
             block_count: Default::default(),
             wal_count: Default::default(),
@@ -77,7 +81,13 @@ impl Stage {
 
         let block = MultiEraBlock::decode(body).or_panic()?;
 
-        crate::state::apply_block_batch([&block], &self.ledger, &self.genesis).or_panic()?;
+        crate::state::apply_block_batch(
+            [&block],
+            &self.ledger,
+            &self.genesis,
+            self.max_ledger_history,
+        )
+        .or_panic()?;
 
         self.mempool.apply_block(&block);
 

src/sync/mod.rs

@@ -48,6 +48,7 @@ fn define_gasket_policy(config: &Option<gasket::retries::Policy>) -> gasket::run
 pub fn pipeline(
     config: &Config,
     upstream: &UpstreamConfig,
+    storage: &StorageConfig,
     wal: WalStore,
     ledger: LedgerStore,
     genesis: Arc<Genesis>,
@@ -65,7 +66,13 @@ pub fn pipeline(
 
     let mut roll = roll::Stage::new(wal.clone());
 
-    let mut apply = apply::Stage::new(wal.clone(), ledger, mempool.clone(), genesis);
+    let mut apply = apply::Stage::new(
+        wal.clone(),
+        ledger,
+        mempool.clone(),
+        genesis,
+        storage.max_ledger_history,
+    );
 
     let submit = submit::Stage::new(
         upstream.peer_address.clone(),