feat: make historical epochs state

Cargo.toml

@@ -24,6 +24,7 @@ members = [
     "modules/accounts_state",               # Tracks stake and reward accounts
     "modules/assets_state",                 # Tracks native asset mints and burns
     "modules/historical_accounts_state",    # Tracks historical account information
+    "modules/historical_epochs_state",      # Tracks historical epochs information
     "modules/consensus",                    # Chooses favoured chain across multiple options
     "modules/chain_store",                  # Tracks historical information about blocks and TXs
     "modules/tx_submitter",                 # Submits TXs to peers

common/Cargo.toml

@@ -26,7 +26,7 @@ hex = { workspace = true }
 memmap2 = "0.9"
 num-rational = { version = "0.4.2", features = ["serde"] }
 regex = "1"
-serde = { workspace = true }
+serde = { workspace = true, features = ["rc"] }
 serde_json = { workspace = true }
 serde_with = { workspace = true, features = ["base64"] }
 tempfile = "3"

common/src/cbor.rs

@@ -0,0 +1,95 @@
+// Custom codec module for u128 using CBOR bignum encoding
+pub mod u128_cbor_codec {
+    use minicbor::{Decoder, Encoder};
+
+    /// Encode u128 as CBOR Tag 2 (positive bignum)
+    /// For use with `#[cbor(with = "u128_cbor_codec")]`
+    pub fn encode<C, W: minicbor::encode::Write>(
+        v: &u128,
+        e: &mut Encoder<W>,
+        _ctx: &mut C,
+    ) -> Result<(), minicbor::encode::Error<W::Error>> {
+        // Tag 2 = positive bignum
+        e.tag(minicbor::data::Tag::new(2))?;
+
+        // Optimize: only encode non-zero leading bytes
+        let bytes = v.to_be_bytes();
+        let first_nonzero = bytes.iter().position(|&b| b != 0).unwrap_or(15);
+        e.bytes(&bytes[first_nonzero..])?;
+        Ok(())
+    }
+
+    /// Decode u128 from CBOR Tag 2 (positive bignum)
+    pub fn decode<'b, C>(
+        d: &mut Decoder<'b>,
+        _ctx: &mut C,
+    ) -> Result<u128, minicbor::decode::Error> {
+        // Expect Tag 2
+        let tag = d.tag()?;
+        if tag != minicbor::data::Tag::new(2) {
+            return Err(minicbor::decode::Error::message(
+                "Expected CBOR Tag 2 (positive bignum) for u128",
+            ));
+        }
+
+        let bytes = d.bytes()?;
+        if bytes.len() > 16 {
+            return Err(minicbor::decode::Error::message(
+                "Bignum too large for u128 (max 16 bytes)",
+            ));
+        }
+
+        // Pad with leading zeros to make 16 bytes (big-endian)
+        let mut arr = [0u8; 16];
+        arr[16 - bytes.len()..].copy_from_slice(bytes);
+        Ok(u128::from_be_bytes(arr))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::u128_cbor_codec;
+    use minicbor::{Decode, Encode};
+
+    #[derive(Debug, PartialEq, Encode, Decode)]
+    struct TestStruct {
+        #[cbor(n(0), with = "u128_cbor_codec")]
+        value: u128,
+    }
+
+    #[test]
+    fn test_u128_zero() {
+        let original = TestStruct { value: 0 };
+        let encoded = minicbor::to_vec(&original).unwrap();
+        let decoded: TestStruct = minicbor::decode(&encoded).unwrap();
+        assert_eq!(original, decoded);
+    }
+
+    #[test]
+    fn test_u128_max() {
+        let original = TestStruct { value: u128::MAX };
+        let encoded = minicbor::to_vec(&original).unwrap();
+        let decoded: TestStruct = minicbor::decode(&encoded).unwrap();
+        assert_eq!(original, decoded);
+    }
+
+    #[test]
+    fn test_u128_boundary_values() {
+        let test_values = [
+            0u128,
+            1,
+            127,                    // Max 1-byte value
+            u64::MAX as u128,       // 18446744073709551615
+            (u64::MAX as u128) + 1, // First value needing >64 bits
+            u128::MAX - 1,          // Near max
+            u128::MAX,              // Maximum u128 value
+        ];
+
+        for &val in &test_values {
+            let original = TestStruct { value: val };
+            let encoded = minicbor::to_vec(&original).unwrap();
+            let decoded: TestStruct = minicbor::decode(&encoded).unwrap();
+            assert_eq!(original, decoded, "Failed for value {}", val);
+        }
+    }
+}

common/src/lib.rs

@@ -2,6 +2,7 @@
 
 pub mod address;
 pub mod calculations;
+pub mod cbor;
 pub mod cip19;
 pub mod commands;
 pub mod crypto;

common/src/messages.rs

@@ -26,6 +26,7 @@ use crate::queries::{
     transactions::{TransactionsStateQuery, TransactionsStateQueryResponse},
 };
 
+use crate::cbor::u128_cbor_codec;
 use crate::types::*;
 use crate::validation::ValidationStatus;
 
@@ -141,47 +142,68 @@ pub struct BlockTxsMessage {
 }
 
 /// Epoch activity - sent at end of epoch
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(
+    Debug,
+    Clone,
+    serde::Serialize,
+    serde::Deserialize,
+    minicbor::Encode,
+    minicbor::Decode,
+    PartialEq,
+)]
 pub struct EpochActivityMessage {
     /// Epoch which has ended
+    #[n(0)]
     pub epoch: u64,
 
     /// Epoch start time
     /// UNIX timestamp
+    #[n(1)]
     pub epoch_start_time: u64,
 
     /// Epoch end time
     /// UNIX timestamp
+    #[n(2)]
     pub epoch_end_time: u64,
 
     /// When first block of this epoch was created
+    #[n(3)]
     pub first_block_time: u64,
 
     /// Block height of first block of this epoch
+    #[n(4)]
     pub first_block_height: u64,
 
     /// When last block of this epoch was created
+    #[n(5)]
     pub last_block_time: u64,
 
     /// Block height of last block of this epoch
+    #[n(6)]
     pub last_block_height: u64,
 
     /// Total blocks in this epoch
+    #[n(7)]
     pub total_blocks: usize,
 
     /// Total txs in this epoch
+    #[n(8)]
     pub total_txs: u64,
 
     /// Total outputs of all txs in this epoch
+    #[cbor(n(9), with = "u128_cbor_codec")]
     pub total_outputs: u128,
 
     /// Total fees in this epoch
+    #[n(10)]
     pub total_fees: u64,
 
     /// Map of SPO IDs to blocks produced
+    #[n(11)]
     pub spo_blocks: Vec<(PoolId, usize)>,
 
     /// Nonce
+    #[n(12)]
     pub nonce: Option<Nonce>,
 }
 

common/src/protocol_params.rs

@@ -254,21 +254,46 @@ impl ProtocolVersion {
 }
 
 #[derive(
-    Default, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize,
+    Default,
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    serde::Serialize,
+    serde::Deserialize,
+    minicbor::Encode,
+    minicbor::Decode,
 )]
 #[serde(rename_all = "PascalCase")]
 pub enum NonceVariant {
+    #[n(0)]
     #[default]
     NeutralNonce,
+    #[n(1)]
     Nonce,
 }
 
 pub type NonceHash = [u8; 32];
 
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)]
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    serde::Serialize,
+    serde::Deserialize,
+    minicbor::Encode,
+    minicbor::Decode,
+)]
 #[serde(rename_all = "camelCase")]
 pub struct Nonce {
+    #[n(0)]
     pub tag: NonceVariant,
+    #[n(1)]
     pub hash: Option<NonceHash>,
 }
 

common/src/queries/epochs.rs

@@ -4,12 +4,20 @@ use crate::{messages::EpochActivityMessage, protocol_params::ProtocolParams, Poo
 pub const DEFAULT_EPOCHS_QUERY_TOPIC: (&str, &str) =
     ("epochs-state-query-topic", "cardano.query.epochs");
 
+pub const DEFAULT_HISTORICAL_EPOCHS_QUERY_TOPIC: (&str, &str) = (
+    "historical-epochs-state-query-topic",
+    "cardano.query.historical.epochs",
+);
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum EpochsStateQuery {
     GetLatestEpoch,
+
+    // Served from historical epochs state
     GetEpochInfo { epoch_number: u64 },
     GetNextEpochs { epoch_number: u64 },
     GetPreviousEpochs { epoch_number: u64 },
+
     GetEpochStakeDistribution { epoch_number: u64 },
     GetEpochStakeDistributionByPool { epoch_number: u64 },
     GetLatestEpochBlocksMintedByPool { spo_id: PoolId },

modules/accounts_state/src/accounts_state.rs

@@ -51,7 +51,7 @@ const DEFAULT_SPO_REWARDS_TOPIC: &str = "cardano.spo.rewards";
 const DEFAULT_PROTOCOL_PARAMETERS_TOPIC: &str = "cardano.protocol.parameters";
 const DEFAULT_STAKE_REWARD_DELTAS_TOPIC: &str = "cardano.stake.reward.deltas";
 
-const DEFAULT_SPDD_DB_PATH: (&str, &str) = ("spdd-db-path", "./spdd_db");
+const DEFAULT_SPDD_DB_PATH: (&str, &str) = ("spdd-db-path", "./fjall-spdd");
 const DEFAULT_SPDD_RETENTION_EPOCHS: (&str, u64) = ("spdd-retention-epochs", 0);
 
 /// Accounts State module
@@ -403,24 +403,28 @@ impl AccountsState {
         let parameters_topic = config
             .get_string("protocol-parameters-topic")
             .unwrap_or(DEFAULT_PROTOCOL_PARAMETERS_TOPIC.to_string());
+        info!("Creating protocol parameters subscriber on '{parameters_topic}'");
 
         // Publishing topics
         let drep_distribution_topic = config
             .get_string("publish-drep-distribution-topic")
             .unwrap_or(DEFAULT_DREP_DISTRIBUTION_TOPIC.to_string());
+        info!("Creating DRep distribution publisher on '{drep_distribution_topic}'");
 
         let spo_distribution_topic = config
             .get_string("publish-spo-distribution-topic")
             .unwrap_or(DEFAULT_SPO_DISTRIBUTION_TOPIC.to_string());
+        info!("Creating SPO distribution publisher on '{spo_distribution_topic}'");
 
         let spo_rewards_topic = config
             .get_string("publish-spo-rewards-topic")
             .unwrap_or(DEFAULT_SPO_REWARDS_TOPIC.to_string());
+        info!("Creating SPO rewards publisher on '{spo_rewards_topic}'");
 
         let stake_reward_deltas_topic = config
             .get_string("publish-stake-reward-deltas-topic")
             .unwrap_or(DEFAULT_STAKE_REWARD_DELTAS_TOPIC.to_string());
-        info!("Creating stake reward deltas subscriber on '{stake_reward_deltas_topic}'");
+        info!("Creating stake reward deltas publisher on '{stake_reward_deltas_topic}'");
 
         let spdd_db_path =
             config.get_string(DEFAULT_SPDD_DB_PATH.0).unwrap_or(DEFAULT_SPDD_DB_PATH.1.to_string());

modules/address_state/.gitignore

@@ -1 +1,2 @@
-db/
+# fjall immutable db
+fjall-*/

modules/address_state/src/address_state.rs

@@ -32,7 +32,7 @@ const DEFAULT_PARAMETERS_SUBSCRIBE_TOPIC: (&str, &str) =
     ("parameters-subscribe-topic", "cardano.protocol.parameters");
 
 // Configuration defaults
-const DEFAULT_ADDRESS_DB_PATH: (&str, &str) = ("db-path", "./db");
+const DEFAULT_ADDRESS_DB_PATH: (&str, &str) = ("db-path", "./fjall-addresses");
 const DEFAULT_CLEAR_ON_START: (&str, bool) = ("clear-on-start", true);
 const DEFAULT_STORE_INFO: (&str, bool) = ("store-info", false);
 const DEFAULT_STORE_TOTALS: (&str, bool) = ("store-totals", false);

modules/address_state/src/state.rs

@@ -1,8 +1,4 @@
-use std::{
-    collections::HashSet,
-    path::{Path, PathBuf},
-    sync::Arc,
-};
+use std::{collections::HashSet, path::Path, sync::Arc};
 
 use acropolis_common::{
     Address, AddressDelta, AddressTotals, BlockInfo, ShelleyAddress, TxIdentifier, TxTotals,
@@ -55,13 +51,8 @@ pub struct State {
 
 impl State {
     pub async fn new(config: &AddressStorageConfig) -> Result<Self> {
-        let db_path = if Path::new(&config.db_path).is_relative() {
-            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(&config.db_path)
-        } else {
-            PathBuf::from(&config.db_path)
-        };
-
-        let store = Arc::new(ImmutableAddressStore::new(&db_path, config.clear_on_start)?);
+        let db_path = Path::new(&config.db_path);
+        let store = Arc::new(ImmutableAddressStore::new(db_path, config.clear_on_start)?);
 
         let mut config = config.clone();
         config.skip_until = store.get_last_epoch_stored().await?;

modules/epochs_state/Cargo.toml

@@ -15,7 +15,6 @@ caryatid_sdk = { workspace = true }
 
 anyhow = { workspace = true }
 config = { workspace = true }
-dashmap = { workspace = true }
 hex = { workspace = true }
 imbl = { workspace = true }
 pallas = { workspace = true }