feat: snapshot bootstrapper module

Author: yHSJ

Cargo.toml

@@ -6,23 +6,24 @@ members = [
     "common",
 
     # Modules
-    "modules/genesis_bootstrapper",    # Genesis bootstrap UTXOs
-    "modules/mithril_snapshot_fetcher",# Mithril snapshot fetcher
-    "modules/upstream_chain_fetcher",  # Upstream chain fetcher
-    "modules/block_unpacker",          # Block to transaction unpacker
-    "modules/tx_unpacker",             # Tx to UTXO unpacker
-    "modules/utxo_state",              # UTXO state
-    "modules/spo_state",               # SPO state
-    "modules/drep_state",              # DRep state
-    "modules/governance_state",        # Governance state
-    "modules/parameters_state",        # Keeps track of protocol parameters
-    "modules/stake_delta_filter",      # Filters address deltas
-    "modules/epoch_activity_counter",  # Counts fees and block producers for rewards
-    "modules/accounts_state",          # Tracks stake and reward accounts
+    "modules/genesis_bootstrapper",     # Genesis bootstrap UTXOs
+    "modules/mithril_snapshot_fetcher", # Mithril snapshot fetcher
+    "modules/snapshot_bootstrapper",    # Bootstrap state from a ledger snapshot
+    "modules/upstream_chain_fetcher",   # Upstream chain fetcher
+    "modules/block_unpacker",           # Block to transaction unpacker
+    "modules/tx_unpacker",              # Tx to UTXO unpacker
+    "modules/utxo_state",               # UTXO state
+    "modules/spo_state",                # SPO state
+    "modules/drep_state",               # DRep state
+    "modules/parameters_state",         # Keeps track of protocol parameters
+    "modules/governance_state",         # Governance state
+    "modules/stake_delta_filter",       # Filters address deltas
+    "modules/epoch_activity_counter",   # Counts fees and block producers for rewards
+    "modules/accounts_state",           # Tracks stake and reward accounts
 
     # Process builds
-    "processes/omnibus",               # All-inclusive omnibus process
-    "processes/replayer",              # All-inclusive process to replay messages
+    "processes/omnibus",  # All-inclusive omnibus process
+    "processes/replayer", # All-inclusive process to replay messages
 ]
 
 resolver = "2"

common/Cargo.toml

@@ -26,6 +26,8 @@ serde_json = "1.0"
 serde_with = { version = "3.12.0", features = ["hex"] }
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1.40"
+minicbor = { version = "0.26.0", features = ["std", "half", "derive"] }
+
 
 [lib]
 crate-type = ["rlib"]

common/src/ledger_state.rs

@@ -0,0 +1,98 @@
+use crate::{KeyHash, PoolRegistration};
+use anyhow::{bail, Context, Result};
+use std::{collections::BTreeMap, fs, path::Path};
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
+pub struct LedgerState {
+    pub spo_state: SPOState,
+}
+
+pub struct UTxOState {}
+
+pub struct StakeDistributionState {}
+
+pub struct AccountState {}
+
+pub struct ParametersState {}
+
+#[derive(
+    Debug, Clone, serde::Serialize, serde::Deserialize, minicbor::Decode, minicbor::Encode, Default,
+)]
+pub struct SPOState {
+    #[n(0)]
+    pub pools: BTreeMap<KeyHash, PoolRegistration>,
+    #[n(1)]
+    pub retiring: BTreeMap<KeyHash, u64>,
+}
+
+pub struct DRepState {}
+
+pub struct ProposalState {}
+
+pub struct VotingState {}
+
+impl LedgerState {
+    pub fn from_directory(directory_path: impl AsRef<Path>) -> Result<Self> {
+        let directory_path = directory_path.as_ref();
+        if !directory_path.exists() {
+            bail!("directory does not exist: {:?}", directory_path);
+        }
+
+        if !directory_path.is_dir() {
+            bail!("path is not a directory: {:?}", directory_path);
+        }
+
+        let mut ledger_state = Self::default();
+        ledger_state
+            .load_from_directory(directory_path)
+            .with_context(|| {
+                format!(
+                    "Failed to load ledger state from directory: {:?}",
+                    directory_path
+                )
+            })?;
+
+        Ok(ledger_state)
+    }
+
+    fn load_from_directory(&mut self, directory_path: impl AsRef<Path>) -> Result<()> {
+        let directory_path = directory_path.as_ref();
+        let entries = fs::read_dir(directory_path)
+            .with_context(|| format!("failed to read directory: {:?}", directory_path))?;
+
+        for entry in entries {
+            let entry = entry.with_context(|| "failed to read directory entry")?;
+            let path = entry.path();
+
+            if path.is_file() && path.extension().map_or(false, |ext| ext == "cbor") {
+                self.load_cbor_file(&path)
+                    .with_context(|| format!("failed to load CBOR file: {:?}", path))?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn load_cbor_file(&mut self, file_path: impl AsRef<Path>) -> Result<()> {
+        let file_path = file_path.as_ref();
+        let filename = file_path
+            .file_stem()
+            .and_then(|s| s.to_str())
+            .with_context(|| format!("invalid filename: {:?}", file_path))?;
+
+        let bytes =
+            fs::read(file_path).with_context(|| format!("failed to read file: {:?}", file_path))?;
+
+        match filename {
+            "pools" => {
+                self.spo_state = minicbor::decode(&bytes)
+                    .with_context(|| format!("failed to decode SPO state from: {:?}", file_path))?;
+            }
+            _ => {
+                // ignore unknown cbor files
+            }
+        }
+
+        Ok(())
+    }
+}

common/src/lib.rs

@@ -4,6 +4,7 @@ pub mod address;
 pub mod calculations;
 pub mod cip19;
 pub mod crypto;
+pub mod ledger_state;
 pub mod messages;
 pub mod params;
 pub mod rational_number;

common/src/messages.rs

@@ -3,6 +3,8 @@
 // We don't use these messages in the acropolis_common crate itself
 #![allow(dead_code)]
 
+use crate::ledger_state::SPOState;
+
 use crate::types::*;
 
 // Caryatid core messages which we re-export
@@ -155,6 +157,11 @@ pub enum CardanoMessage {
     StakeAddressDeltas(StakeAddressDeltasMessage),       // Stake part of address deltas
 }
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub enum SnapshotStateMessage {
+    SPOState(SPOState),
+}
+
 // === Global message enum ===
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum Message {
@@ -171,6 +178,9 @@ pub enum Message {
 
     // Cardano messages with attached BlockInfo
     Cardano((BlockInfo, CardanoMessage)),
+
+    // Initialize state from a snapshot
+    SnapshotState(SnapshotStateMessage),
 }
 
 impl Default for Message {

common/src/types.rs

@@ -8,6 +8,7 @@ use anyhow::anyhow;
 use bech32::{Bech32, Hrp};
 use bitmask_enum::bitmask;
 use hex::decode;
+use minicbor::data::Tag;
 use serde::{Deserialize, Serialize};
 use serde_with::{hex::Hex, serde_as};
 use std::collections::{HashMap, HashSet};
@@ -149,6 +150,33 @@ pub struct Ratio {
     pub denominator: u64,
 }
 
+// NOTE: Implementations from Pallas, why aren't we just using Pallas types here?
+impl<'b, C> minicbor::decode::Decode<'b, C> for Ratio {
+    fn decode(d: &mut minicbor::Decoder<'b>, ctx: &mut C) -> Result<Self, minicbor::decode::Error> {
+        // TODO: Enforce tag == 30 & array of size 2
+        d.tag()?;
+        d.array()?;
+        Ok(Ratio {
+            numerator: d.decode_with(ctx)?,
+            denominator: d.decode_with(ctx)?,
+        })
+    }
+}
+
+impl<C> minicbor::encode::Encode<C> for Ratio {
+    fn encode<W: minicbor::encode::Write>(
+        &self,
+        e: &mut minicbor::Encoder<W>,
+        ctx: &mut C,
+    ) -> Result<(), minicbor::encode::Error<W::Error>> {
+        e.tag(Tag::new(30))?;
+        e.array(2)?;
+        e.encode_with(self.numerator, ctx)?;
+        e.encode_with(self.denominator, ctx)?;
+        Ok(())
+    }
+}
+
 /// General credential
 #[derive(
     Debug, Clone, Ord, Eq, PartialEq, PartialOrd, Hash, serde::Serialize, serde::Deserialize,
@@ -237,53 +265,129 @@ pub enum Relay {
     MultiHostName(MultiHostName),
 }
 
+// NOTE: Implementations from Pallas, why aren't we just using Pallas types here?
+impl<'b, C> minicbor::decode::Decode<'b, C> for Relay {
+    fn decode(d: &mut minicbor::Decoder<'b>, ctx: &mut C) -> Result<Self, minicbor::decode::Error> {
+        d.array()?;
+        let variant = d.u16()?;
+
+        match variant {
+            0 => Ok(Relay::SingleHostAddr(SingleHostAddr {
+                port: d.decode_with(ctx)?,
+                ipv4: d.decode_with(ctx)?,
+                ipv6: d.decode_with(ctx)?,
+            })),
+            1 => Ok(Relay::SingleHostName(SingleHostName {
+                port: d.decode_with(ctx)?,
+                dns_name: d.decode_with(ctx)?,
+            })),
+            2 => Ok(Relay::MultiHostName(MultiHostName {
+                dns_name: d.decode_with(ctx)?,
+            })),
+            _ => Err(minicbor::decode::Error::message(
+                "invalid variant id for Relay",
+            )),
+        }
+    }
+}
+
+impl<C> minicbor::encode::Encode<C> for Relay {
+    fn encode<W: minicbor::encode::Write>(
+        &self,
+        e: &mut minicbor::Encoder<W>,
+        ctx: &mut C,
+    ) -> Result<(), minicbor::encode::Error<W::Error>> {
+        match self {
+            Relay::SingleHostAddr(SingleHostAddr { port, ipv4, ipv6 }) => {
+                e.array(4)?;
+                e.encode_with(0, ctx)?;
+                e.encode_with(port, ctx)?;
+                e.encode_with(ipv4, ctx)?;
+                e.encode_with(ipv6, ctx)?;
+
+                Ok(())
+            }
+            Relay::SingleHostName(SingleHostName { port, dns_name }) => {
+                e.array(3)?;
+                e.encode_with(1, ctx)?;
+                e.encode_with(port, ctx)?;
+                e.encode_with(dns_name, ctx)?;
+
+                Ok(())
+            }
+            Relay::MultiHostName(MultiHostName { dns_name }) => {
+                e.array(2)?;
+                e.encode_with(2, ctx)?;
+                e.encode_with(dns_name, ctx)?;
+
+                Ok(())
+            }
+        }
+    }
+}
+
 /// Pool metadata
 #[serde_as]
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(
+    Debug, Clone, serde::Serialize, serde::Deserialize, minicbor::Encode, minicbor::Decode,
+)]
 pub struct PoolMetadata {
     /// Metadata URL
+    #[n(0)]
     pub url: String,
 
     /// Metadata hash
     #[serde_as(as = "Hex")]
+    #[n(1)]
     pub hash: DataHash,
 }
 
 type RewardAccount = Vec<u8>;
 
 /// Pool registration data
 #[serde_as]
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(
+    Debug, Clone, serde::Serialize, serde::Deserialize, minicbor::Decode, minicbor::Encode,
+)]
 pub struct PoolRegistration {
     /// Operator pool key hash - used as ID
     #[serde_as(as = "Hex")]
+    #[n(0)]
     pub operator: KeyHash,
 
     /// VRF key hash
     #[serde_as(as = "Hex")]
+    #[n(1)]
     pub vrf_key_hash: KeyHash,
 
     /// Pledged Ada
+    #[n(2)]
     pub pledge: Lovelace,
 
     /// Fixed cost
+    #[n(3)]
     pub cost: Lovelace,
 
     /// Marginal cost (fraction)
+    #[n(4)]
     pub margin: Ratio,
 
     /// Reward account
     #[serde_as(as = "Hex")]
+    #[n(5)]
     pub reward_account: Vec<u8>,
 
     /// Pool owners by their key hash
     #[serde_as(as = "Vec<Hex>")]
+    #[n(6)]
     pub pool_owners: Vec<KeyHash>,
 
     // Relays
+    #[n(7)]
     pub relays: Vec<Relay>,
 
     // Metadata
+    #[n(8)]
     pub pool_metadata: Option<PoolMetadata>,
 }
 

modules/snapshot_bootstrapper/Cargo.toml

@@ -0,0 +1,24 @@
+# Acropolis snapshot bootstrapper module
+
+[package]
+name = "acropolis_module_snapshot_bootstrapper"
+version = "0.1.0"
+edition = "2021"
+authors = ["Josh Marchand <[email protected]>"]
+description = "Snapshot bootstrapper Caryatid module for Acropolis"
+license = "Apache-2.0"
+
+[dependencies]
+caryatid_sdk = "0.9.0"
+acropolis_common = { path = "../../common" }
+hex = "0.4"
+fraction = "0.15"
+pallas = "0.32.0"
+anyhow = "1.0"
+tokio = { version = "1", features = ["full"] }
+config = "0.15.11"
+tracing = "0.1.40"
+serde_json = "1.0.138"
+
+[lib]
+path = "src/snapshot_bootstrapper.rs"