refactor: use Redb for ledger data (#199)

Cargo.toml

@@ -12,8 +12,8 @@ authors = ["Santiago Carmuega <santiago@carmuega.me>"]
 
 
 [dependencies]
-# pallas = { git = "https://github.com/txpipe/pallas.git", features = ["unstable"] }
-pallas = { version = "^0.25", features = ["unstable"] }
+pallas = { git = "https://github.com/txpipe/pallas.git", features = ["unstable"] }
+# pallas = { version = "^0.25", features = ["unstable"] }
 # pallas = { path = "../pallas/pallas", features = ["unstable"] }
 
 gasket = { version = "^0.5", features = ["derive"] }
@@ -49,6 +49,7 @@ async-stream = "0.3.5"
 serde_with = "3.4.0"
 mithril-client = { version = "0.5.17", optional = true, features = ["fs"] }
 protoc-wkt = "1.0.0"
+itertools = "0.12.1"
 
 [dev-dependencies]
 tempfile = "3.3.0"

src/bin/dolos/bootstrap.rs

@@ -165,18 +165,22 @@ pub fn run(config: &super::Config, args: &Args) -> miette::Result<()> {
     let (mut wal, mut chain, mut ledger) = empty_stores.unwrap();
 
     ledger
-        .apply_origin(&byron_genesis)
+        .apply(&[dolos::ledger::compute_origin_delta(&byron_genesis)])
         .into_diagnostic()
         .context("applying origin utxos")?;
 
     for block in iter {
         let block = match block {
-            Ok(x) => x,
+            Ok(x) if x.is_empty() => {
+                warn!("can't continue reading from immutable db");
+                break;
+            }
             Err(err) => {
                 dbg!(err);
                 warn!("can't continue reading from immutable db");
                 break;
             }
+            Ok(x) => x,
         };
 
         let blockd = MultiEraBlock::decode(&block)
@@ -191,7 +195,7 @@ pub fn run(config: &super::Config, args: &Args) -> miette::Result<()> {
             .context("adding chain entry")?;
 
         ledger
-            .apply_block(&blockd)
+            .apply(&[dolos::ledger::compute_delta(&blockd)])
             .into_diagnostic()
             .context("applyting ledger block")?;
     }

src/bin/dolos/common.rs

@@ -3,7 +3,7 @@ use std::path::Path;
 use tracing::Level;
 use tracing_subscriber::{filter::Targets, prelude::*};
 
-use dolos::{prelude::*, storage::applydb::ApplyDB};
+use dolos::{ledger::store::LedgerStore, prelude::*};
 
 use crate::LoggingConfig;
 
@@ -15,7 +15,7 @@ fn define_rolldb_path(config: &crate::Config) -> &Path {
         .unwrap_or_else(|| Path::new("/rolldb"))
 }
 
-pub type Stores = (wal::Store, chain::Store, ApplyDB);
+pub type Stores = (wal::Store, chain::Store, LedgerStore);
 
 pub fn open_data_stores(config: &crate::Config) -> Result<Stores, Error> {
     let rolldb_path = define_rolldb_path(config);
@@ -29,7 +29,7 @@ pub fn open_data_stores(config: &crate::Config) -> Result<Stores, Error> {
 
     let chain = chain::Store::open(rolldb_path.join("chain")).map_err(Error::storage)?;
 
-    let ledger = ApplyDB::open(rolldb_path.join("ledger")).map_err(Error::storage)?;
+    let ledger = LedgerStore::open(rolldb_path.join("ledger")).map_err(Error::storage)?;
 
     Ok((wal, chain, ledger))
 }
@@ -40,7 +40,7 @@ pub fn destroy_data_stores(config: &crate::Config) -> Result<(), Error> {
 
     wal::Store::destroy(rolldb_path.join("wal")).map_err(Error::storage)?;
     chain::Store::destroy(rolldb_path.join("chain")).map_err(Error::storage)?;
-    ApplyDB::destroy(rolldb_path.join("ledger")).map_err(Error::storage)?;
+    //LedgerStore::destroy(rolldb_path.join("ledger")).map_err(Error::storage)?;
 
     Ok(())
 }

src/bin/dolos/data.rs

@@ -1,8 +1,13 @@
 use std::path::Path;
 
+use dolos::ledger::{pparams::Genesis, ChainPoint, PParamsBody};
+use itertools::Itertools;
 use miette::IntoDiagnostic;
 
-use pallas::{ledger::traverse::MultiEraBlock, storage::rolldb::chain};
+use pallas::{
+    ledger::traverse::{Era, MultiEraBlock, MultiEraUpdate},
+    storage::rolldb::chain,
+};
 
 #[allow(dead_code)]
 fn dump_txs(chain: &chain::Store) -> miette::Result<()> {
@@ -49,13 +54,41 @@ pub fn run(config: &super::Config, _args: &Args) -> miette::Result<()> {
 
     println!("---");
 
-    if let Some((slot, hash)) = ledger.cursor().unwrap() {
+    if let Some(ChainPoint(slot, hash)) = ledger.cursor().unwrap() {
         println!("found ledger tip");
         println!("slot: {slot}, hash: {hash}");
     } else {
         println!("chain is empty");
     }
 
+    println!("---");
+
+    let byron = pallas::ledger::configs::byron::from_file(&config.byron.path).unwrap();
+    let shelley = pallas::ledger::configs::shelley::from_file(&config.shelley.path).unwrap();
+    let alonzo = pallas::ledger::configs::alonzo::from_file(&config.alonzo.path).unwrap();
+
+    let data: Vec<_> = ledger.get_pparams(46153027).into_diagnostic()?;
+
+    let updates = data
+        .iter()
+        .map(|PParamsBody(era, cbor)| -> miette::Result<MultiEraUpdate> {
+            let era = Era::try_from(*era).into_diagnostic()?;
+            MultiEraUpdate::decode_for_era(era, cbor).into_diagnostic()
+        })
+        .try_collect()?;
+
+    let merged = dolos::ledger::pparams::fold_pparams(
+        Genesis {
+            byron: &byron,
+            shelley: &shelley,
+            alonzo: &alonzo,
+        },
+        updates,
+        500,
+    );
+
+    dbg!(merged);
+
     // WIP utility to dump tx data for debugging purposes. Should be implemented as
     // a subcommand.