feat: update epoch state to handle epoch times, total txs, and total outputs

- Renamed BlockFeesMessage to BlockTxsMessage and added fields for total transactions and total output.
- Enhanced epoch's info endpoint with new fields for epoch start time, end time, first and last block times, total transactions, and total outputs.
- Updated state management to handle new transaction metrics during epoch processing.
- Adjusted related tests and configurations to reflect changes in message structures and topics.

Author: golddydev

common/src/messages.rs

@@ -123,7 +123,13 @@ pub struct StakeRewardDeltasMessage {
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct BlockFeesMessage {
+pub struct BlockTxsMessage {
+    /// Total transactions
+    pub total_txs: u64,
+
+    /// Total output
+    pub total_output: u128,
+
     /// Total fees
     pub total_fees: u64,
 }
@@ -134,9 +140,27 @@ pub struct EpochActivityMessage {
     /// Epoch which has ended
     pub epoch: u64,
 
+    /// Epoch start time
+    pub epoch_start_time: u64,
+
+    /// Epoch end time
+    pub epoch_end_time: u64,
+
+    /// First block time
+    pub first_block_time: u64,
+
+    /// Last block time
+    pub last_block_time: u64,
+
     /// Total blocks in this epoch
     pub total_blocks: usize,
 
+    /// Total txs in this epoch
+    pub total_txs: u64,
+
+    /// Total outputs of all txs in this epoch
+    pub total_outputs: u128,
+
     /// Total fees in this epoch
     pub total_fees: u64,
 
@@ -252,10 +276,10 @@ pub enum CardanoMessage {
     AddressDeltas(AddressDeltasMessage),     // Address deltas received
     Withdrawals(WithdrawalsMessage),         // Withdrawals from reward accounts
     PotDeltas(PotDeltasMessage),             // Changes to pot balances
-    BlockFees(BlockFeesMessage),             // Total fees in a block
-    EpochActivity(EpochActivityMessage),     // Total fees and VRF keys for an epoch
-    DRepState(DRepStateMessage),             // Active DReps at epoch end
-    SPOState(SPOStateMessage),               // Active SPOs at epoch end
+    BlockTxs(BlockTxsMessage), // Transactions Info like total count, total output, total fees in a block
+    EpochActivity(EpochActivityMessage), // Total fees and VRF keys for an epoch
+    DRepState(DRepStateMessage), // Active DReps at epoch end
+    SPOState(SPOStateMessage), // Active SPOs at epoch end
     GovernanceProcedures(GovernanceProceduresMessage), // Governance procedures received
 
     // Protocol Parameters

common/src/params.rs

@@ -1,2 +1,3 @@
 pub const SECURITY_PARAMETER_K: u64 = 2160;
 pub const TECHNICAL_PARAMETER_POOL_RETIRE_MAX_EPOCH: u64 = 18;
+pub const EPOCH_LENGTH: u64 = 432000;

modules/epochs_state/src/epochs_history.rs

@@ -91,7 +91,13 @@ mod tests {
             &block,
             &EpochActivityMessage {
                 epoch: 199,
+                epoch_start_time: 0,
+                epoch_end_time: 0,
+                first_block_time: 0,
+                last_block_time: 0,
                 total_blocks: 1,
+                total_txs: 1,
+                total_outputs: 100,
                 total_fees: 50,
                 vrf_vkey_hashes: vec![],
                 nonce: None,

modules/epochs_state/src/epochs_state.rs

@@ -16,7 +16,7 @@ use config::Config;
 use pallas::ledger::traverse::MultiEraHeader;
 use std::sync::Arc;
 use tokio::sync::Mutex;
-use tracing::{error, info, info_span, Instrument};
+use tracing::{error, info, info_span};
 
 mod epoch_activity_publisher;
 mod epochs_history;
@@ -35,12 +35,13 @@ const DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC: (&str, &str) = (
 );
 const DEFAULT_BLOCK_HEADER_SUBSCRIBE_TOPIC: (&str, &str) =
     ("block-header-subscribe-topic", "cardano.block.header");
-const DEFAULT_BLOCK_FEES_SUBSCRIBE_TOPIC: (&str, &str) =
-    ("block-fees-subscribe-topic", "cardano.block.fees");
+const DEFAULT_BLOCK_TXS_SUBSCRIBE_TOPIC: (&str, &str) =
+    ("block-txs-subscribe-topic", "cardano.block.txs");
 const DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC: (&str, &str) = (
     "protocol-parameters-subscribe-topic",
     "cardano.protocol.parameters",
 );
+
 const DEFAULT_EPOCH_ACTIVITY_PUBLISH_TOPIC: (&str, &str) =
     ("epoch-activity-publish-topic", "cardano.epoch.activity");
 
@@ -59,7 +60,7 @@ impl EpochsState {
         epochs_history: EpochsHistoryState,
         mut bootstrapped_subscription: Box<dyn Subscription<Message>>,
         mut headers_subscription: Box<dyn Subscription<Message>>,
-        mut fees_subscription: Box<dyn Subscription<Message>>,
+        mut block_txs_subscription: Box<dyn Subscription<Message>>,
         mut protocol_parameters_subscription: Box<dyn Subscription<Message>>,
         mut epoch_activity_publisher: EpochActivityPublisher,
     ) -> Result<()> {
@@ -76,12 +77,12 @@ impl EpochsState {
 
         loop {
             // Get a mutable state
-            let mut state = history.lock().await.get_or_init_with(|| State::new());
+            let mut state = history.lock().await.get_or_init_with(|| State::new(&genesis));
             let mut current_block: Option<BlockInfo> = None;
 
             // Read both topics in parallel
             let headers_message_f = headers_subscription.read();
-            let fees_message_f = fees_subscription.read();
+            let block_txs_message_f = block_txs_subscription.read();
 
             // Handle headers first
             let (_, message) = headers_message_f.await?;
@@ -166,18 +167,16 @@ impl EpochsState {
                 _ => error!("Unexpected message type: {message:?}"),
             }
 
-            // Handle block fees second so new epoch's fees don't get counted in the last one
-            let (_, message) = fees_message_f.await?;
+            // Handle block txs second so new epoch's state don't get counted in the last one
+            let (_, message) = block_txs_message_f.await?;
             match message.as_ref() {
-                Message::Cardano((block_info, CardanoMessage::BlockFees(fees_msg))) => {
+                Message::Cardano((block_info, CardanoMessage::BlockTxs(txs_msg))) => {
                     let span =
-                        info_span!("epochs_state.handle_block_fees", block = block_info.number);
-                    async {
+                        info_span!("epochs_state.handle_block_txs", block = block_info.number);
+                    span.in_scope(|| {
                         Self::check_sync(&current_block, &block_info);
-                        state.handle_fees(&block_info, fees_msg.total_fees);
-                    }
-                    .instrument(span)
-                    .await;
+                        state.handle_block_txs(&block_info, txs_msg);
+                    });
                 }
 
                 _ => error!("Unexpected message type: {message:?}"),
@@ -203,10 +202,10 @@ impl EpochsState {
             .unwrap_or(DEFAULT_BLOCK_HEADER_SUBSCRIBE_TOPIC.1.to_string());
         info!("Creating subscriber for headers on '{block_headers_subscribe_topic}'");
 
-        let block_fees_subscribe_topic = config
-            .get_string(DEFAULT_BLOCK_FEES_SUBSCRIBE_TOPIC.0)
-            .unwrap_or(DEFAULT_BLOCK_FEES_SUBSCRIBE_TOPIC.1.to_string());
-        info!("Creating subscriber for fees on '{block_fees_subscribe_topic}'");
+        let block_txs_subscribe_topic = config
+            .get_string(DEFAULT_BLOCK_TXS_SUBSCRIBE_TOPIC.0)
+            .unwrap_or(DEFAULT_BLOCK_TXS_SUBSCRIBE_TOPIC.1.to_string());
+        info!("Creating subscriber for block txs on '{block_txs_subscribe_topic}'");
 
         let protocol_parameters_subscribe_topic = config
             .get_string(DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC.0)
@@ -242,9 +241,9 @@ impl EpochsState {
         // Subscribe
         let bootstrapped_subscription = context.subscribe(&bootstrapped_subscribe_topic).await?;
         let headers_subscription = context.subscribe(&block_headers_subscribe_topic).await?;
-        let fees_subscription = context.subscribe(&block_fees_subscribe_topic).await?;
         let protocol_parameters_subscription =
             context.subscribe(&protocol_parameters_subscribe_topic).await?;
+        let block_txs_subscription = context.subscribe(&block_txs_subscribe_topic).await?;
 
         // Publisher
         let epoch_activity_publisher =
@@ -306,7 +305,7 @@ impl EpochsState {
                 epochs_history,
                 bootstrapped_subscription,
                 headers_subscription,
-                fees_subscription,
+                block_txs_subscription,
                 protocol_parameters_subscription,
                 epoch_activity_publisher,
             )