Merge pull request #244 from input-output-hk/golddydev/epochs-blocks

whankinsiv · web-flow · commit d72276965dc4 · 2025-10-13T16:42:57.000-07:00
feat: epoch blocks endpoints
diff --git a/common/src/messages.rs b/common/src/messages.rs
@@ -148,17 +148,25 @@ pub struct EpochActivityMessage {
     pub epoch: u64,
 
     /// Epoch start time
+    /// UNIX timestamp
     pub epoch_start_time: u64,
 
     /// Epoch end time
+    /// UNIX timestamp
     pub epoch_end_time: u64,
 
-    /// First block time
+    /// When first block of this epoch was created
     pub first_block_time: u64,
 
-    /// Last block time
+    /// Block height of first block of this epoch
+    pub first_block_height: u64,
+
+    /// When last block of this epoch was created
     pub last_block_time: u64,
 
+    /// Block height of last block of this epoch
+    pub last_block_height: u64,
+
     /// Total blocks in this epoch
     pub total_blocks: usize,
 
diff --git a/common/src/queries/epochs.rs b/common/src/queries/epochs.rs
@@ -11,8 +11,6 @@ pub enum EpochsStateQuery {
     GetPreviousEpochs { epoch_number: u64 },
     GetEpochStakeDistribution { epoch_number: u64 },
     GetEpochStakeDistributionByPool { epoch_number: u64 },
-    GetEpochBlockDistribution { epoch_number: u64 },
-    GetEpochBlockDistributionByPool { epoch_number: u64 },
     GetLatestEpochBlocksMintedByPool { vrf_key_hash: KeyHash },
 }
 
@@ -24,8 +22,6 @@ pub enum EpochsStateQueryResponse {
     PreviousEpochs(PreviousEpochs),
     EpochStakeDistribution(EpochStakeDistribution),
     EpochStakeDistributionByPool(EpochStakeDistributionByPool),
-    EpochBlockDistribution(EpochBlockDistribution),
-    EpochBlockDistributionByPool(EpochBlockDistributionByPool),
     LatestEpochBlocksMintedByPool(u64),
 
     NotFound,
@@ -62,9 +58,3 @@ pub struct EpochStakeDistribution {}
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct EpochStakeDistributionByPool {}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct EpochBlockDistribution {}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct EpochBlockDistributionByPool {}
diff --git a/common/src/queries/pools.rs b/common/src/queries/pools.rs
@@ -1,6 +1,6 @@
 use crate::{
-    queries::governance::VoteRecord, rational_number::RationalNumber, BlockHash, KeyHash,
-    PoolEpochState, PoolMetadata, PoolRegistration, PoolRetirement, PoolUpdateEvent, Relay,
+    queries::governance::VoteRecord, rational_number::RationalNumber, KeyHash, PoolEpochState,
+    PoolMetadata, PoolRegistration, PoolRetirement, PoolUpdateEvent, Relay,
 };
 
 pub const DEFAULT_POOLS_QUERY_TOPIC: (&str, &str) =
@@ -41,9 +41,13 @@ pub enum PoolsStateQuery {
     GetPoolTotalBlocksMinted {
         pool_id: KeyHash,
     },
-    GetPoolBlockHashes {
+    GetBlocksByPool {
         pool_id: KeyHash,
     },
+    GetBlocksByPoolAndEpoch {
+        pool_id: KeyHash,
+        epoch: u64,
+    },
     GetPoolUpdates {
         pool_id: KeyHash,
     },
@@ -67,7 +71,10 @@ pub enum PoolsStateQueryResponse {
     PoolRelays(Vec<Relay>),
     PoolDelegators(PoolDelegators),
     PoolTotalBlocksMinted(u64),
-    PoolBlockHashes(Vec<BlockHash>),
+    // Vector of Block Heights
+    BlocksByPool(Vec<u64>),
+    // Vector of Block Heights
+    BlocksByPoolAndEpoch(Vec<u64>),
     PoolUpdates(Vec<PoolUpdateEvent>),
     PoolVotes(Vec<VoteRecord>),
     NotFound,
diff --git a/modules/epochs_state/src/epochs_history.rs b/modules/epochs_state/src/epochs_history.rs
@@ -117,7 +117,9 @@ mod tests {
                 epoch_start_time: 0,
                 epoch_end_time: 0,
                 first_block_time: 0,
+                first_block_height: 0,
                 last_block_time: 0,
+                last_block_height: 0,
                 total_blocks: 1,
                 total_txs: 1,
                 total_outputs: 100,
@@ -147,7 +149,9 @@ mod tests {
                 epoch_start_time: 0,
                 epoch_end_time: 0,
                 first_block_time: 0,
+                first_block_height: 0,
                 last_block_time: 0,
+                last_block_height: 0,
                 total_blocks: 1,
                 total_txs: 1,
                 total_outputs: 100,
@@ -165,7 +169,9 @@ mod tests {
                 epoch_start_time: 0,
                 epoch_end_time: 0,
                 first_block_time: 0,
+                first_block_height: 0,
                 last_block_time: 0,
+                last_block_height: 0,
                 total_blocks: 1,
                 total_txs: 1,
                 total_outputs: 100,
diff --git a/modules/epochs_state/src/state.rs b/modules/epochs_state/src/state.rs
@@ -29,10 +29,16 @@ pub struct State {
     // UNIX timestamp
     first_block_time: u64,
 
+    // first block height
+    first_block_height: u64,
+
     // last block time
     // UNIX timestamp
     last_block_time: u64,
 
+    // last block height
+    last_block_height: u64,
+
     // Map of counts by VRF key hashes
     blocks_minted: HashMap<KeyHash, usize>,
 
@@ -63,7 +69,9 @@ impl State {
             epoch: 0,
             epoch_start_time: genesis.byron_timestamp,
             first_block_time: genesis.byron_timestamp,
+            first_block_height: 0,
             last_block_time: 0,
+            last_block_height: 0,
             blocks_minted: HashMap::new(),
             epoch_blocks: 0,
             epoch_txs: 0,
@@ -178,6 +186,7 @@ impl State {
     // This will update last block time
     pub fn handle_mint(&mut self, block_info: &BlockInfo, vrf_vkey: Option<&[u8]>) {
         self.last_block_time = block_info.timestamp;
+        self.last_block_height = block_info.number;
         self.epoch_blocks += 1;
 
         if let Some(vrf_vkey) = vrf_vkey {
@@ -215,7 +224,9 @@ impl State {
         self.epoch = block_info.epoch;
         self.epoch_start_time = block_info.timestamp;
         self.first_block_time = block_info.timestamp;
+        self.first_block_height = block_info.number;
         self.last_block_time = block_info.timestamp;
+        self.last_block_height = block_info.number;
         self.blocks_minted.clear();
         self.epoch_blocks = 0;
         self.epoch_txs = 0;
@@ -231,7 +242,9 @@ impl State {
             epoch_start_time: self.epoch_start_time,
             epoch_end_time: self.epoch_start_time + EPOCH_LENGTH,
             first_block_time: self.first_block_time,
+            first_block_height: self.first_block_height,
             last_block_time: self.last_block_time,
+            last_block_height: self.last_block_height,
             // NOTE:
             // total_blocks will be missing one
             // This is only because we now ignore EBBs
@@ -422,7 +435,9 @@ mod tests {
         assert!(state.blocks_minted.is_empty());
         assert_eq!(state.epoch_start_time, block.timestamp);
         assert_eq!(state.first_block_time, block.timestamp);
+        assert_eq!(state.first_block_height, block.number);
         assert_eq!(state.last_block_time, block.timestamp);
+        assert_eq!(state.last_block_height, block.number);
 
         let blocks_minted = state.get_latest_epoch_blocks_minted_by_pool(&keyhash(b"vrf_1"));
         assert_eq!(blocks_minted, 0);
diff --git a/modules/rest_blockfrost/src/handlers/epochs.rs b/modules/rest_blockfrost/src/handlers/epochs.rs
@@ -8,9 +8,11 @@ use acropolis_common::{
         accounts::{AccountsStateQuery, AccountsStateQueryResponse},
         epochs::{EpochsStateQuery, EpochsStateQueryResponse},
         parameters::{ParametersStateQuery, ParametersStateQueryResponse},
+        pools::{PoolsStateQuery, PoolsStateQueryResponse},
         spdd::{SPDDStateQuery, SPDDStateQueryResponse},
         utils::query_state,
     },
+    serialization::Bech32WithHrp,
 };
 use anyhow::{anyhow, Result};
 use caryatid_sdk::Context;
@@ -417,9 +419,71 @@ pub async fn handle_epoch_total_blocks_blockfrost(
 }
 
 pub async fn handle_epoch_pool_blocks_blockfrost(
-    _context: Arc<Context<Message>>,
-    _params: Vec<String>,
-    _handlers_config: Arc<HandlersConfig>,
+    context: Arc<Context<Message>>,
+    params: Vec<String>,
+    handlers_config: Arc<HandlersConfig>,
 ) -> Result<RESTResponse> {
-    Ok(RESTResponse::with_text(501, "Not implemented"))
+    if params.len() != 2 {
+        return Ok(RESTResponse::with_text(
+            400,
+            "Expected two parameters: an epoch number and a pool ID",
+        ));
+    }
+    let epoch_number_param = &params[0];
+    let pool_id_param = &params[1];
+
+    let epoch_number = match epoch_number_param.parse::<u64>() {
+        Ok(num) => num,
+        Err(_) => {
+            return Ok(RESTResponse::with_text(
+                400,
+                "Invalid epoch number parameter",
+            ));
+        }
+    };
+
+    let Ok(spo) = Vec::<u8>::from_bech32_with_hrp(pool_id_param, "pool") else {
+        return Ok(RESTResponse::with_text(
+            400,
+            &format!("Invalid Bech32 stake pool ID: {pool_id_param}"),
+        ));
+    };
+
+    // query Pool's Blocks by epoch from spo-state
+    let msg = Arc::new(Message::StateQuery(StateQuery::Pools(
+        PoolsStateQuery::GetBlocksByPoolAndEpoch {
+            pool_id: spo.clone(),
+            epoch: epoch_number,
+        },
+    )));
+
+    let blocks = query_state(
+        &context,
+        &handlers_config.pools_query_topic,
+        msg,
+        |message| match message {
+            Message::StateQueryResponse(StateQueryResponse::Pools(
+                PoolsStateQueryResponse::BlocksByPoolAndEpoch(blocks),
+            )) => Ok(blocks),
+            Message::StateQueryResponse(StateQueryResponse::Pools(
+                PoolsStateQueryResponse::Error(e),
+            )) => Err(anyhow::anyhow!(
+                "Internal server error while retrieving pool block hashes by epoch: {e}"
+            )),
+            _ => Err(anyhow::anyhow!("Unexpected message type")),
+        },
+    )
+    .await?;
+
+    // NOTE:
+    // Need to query chain_store
+    // to get block_hash for each block height
+
+    match serde_json::to_string_pretty(&blocks) {
+        Ok(json) => Ok(RESTResponse::with_json(200, &json)),
+        Err(e) => Ok(RESTResponse::with_text(
+            500,
+            &format!("Internal server error while retrieving pool block hashes by epoch: {e}"),
+        )),
+    }
 }
diff --git a/modules/rest_blockfrost/src/handlers/pools.rs b/modules/rest_blockfrost/src/handlers/pools.rs
@@ -1062,9 +1062,9 @@ pub async fn handle_pool_blocks_blockfrost(
         ));
     };
 
-    // Get block hashes by pool_id from spo_state
+    // Get blocks by pool_id from spo_state
     let pool_blocks_msg = Arc::new(Message::StateQuery(StateQuery::Pools(
-        PoolsStateQuery::GetPoolBlockHashes {
+        PoolsStateQuery::GetBlocksByPool {
             pool_id: spo.clone(),
         },
     )));
@@ -1075,18 +1075,21 @@ pub async fn handle_pool_blocks_blockfrost(
         pool_blocks_msg,
         |message| match message {
             Message::StateQueryResponse(StateQueryResponse::Pools(
-                PoolsStateQueryResponse::PoolBlockHashes(pool_blocks),
+                PoolsStateQueryResponse::BlocksByPool(pool_blocks),
             )) => Ok(pool_blocks),
             Message::StateQueryResponse(StateQueryResponse::Pools(
                 PoolsStateQueryResponse::Error(_),
-            )) => Err(anyhow::anyhow!("Block hashes are not enabled")),
+            )) => Err(anyhow::anyhow!("Blocks are not enabled")),
             _ => Err(anyhow::anyhow!("Unexpected message type")),
         },
     )
     .await?;
 
-    let pool_blocks_rest = pool_blocks.into_iter().map(|b| hex::encode(b)).collect::<Vec<_>>();
-    match serde_json::to_string(&pool_blocks_rest) {
+    // NOTE:
+    // Need to query chain_store
+    // to get block_hash for each block height
+
+    match serde_json::to_string_pretty(&pool_blocks) {
         Ok(json) => Ok(RESTResponse::with_json(200, &json)),
         Err(e) => Ok(RESTResponse::with_text(
             500,
diff --git a/modules/spo_state/src/historical_spo_state.rs b/modules/spo_state/src/historical_spo_state.rs
@@ -1,7 +1,7 @@
 use acropolis_common::{
     queries::governance::VoteRecord, KeyHash, PoolRegistration, PoolUpdateEvent,
 };
-use imbl::HashSet;
+use imbl::{HashSet, OrdMap, Vector};
 use serde::{Deserialize, Serialize};
 
 use crate::store_config::StoreConfig;
@@ -17,6 +17,10 @@ pub struct HistoricalSPOState {
     pub delegators: Option<HashSet<KeyHash>>,
     // SPO's votes
     pub votes: Option<Vec<VoteRecord>>,
+
+    // blocks
+    // <Epoch Number, Block Heights>
+    pub blocks: Option<OrdMap<u64, Vector<u64>>>,
 }
 
 impl HistoricalSPOState {
@@ -27,6 +31,7 @@ impl HistoricalSPOState {
             updates: store_config.store_updates.then(Vec::new),
             delegators: store_config.store_delegators.then(HashSet::new),
             votes: store_config.store_votes.then(Vec::new),
+            blocks: store_config.store_blocks.then(OrdMap::new),
         }
     }
 
@@ -55,4 +60,21 @@ impl HistoricalSPOState {
     pub fn remove_delegator(&mut self, delegator: &KeyHash) -> Option<bool> {
         self.delegators.as_mut().and_then(|delegators| Some(delegators.remove(delegator).is_some()))
     }
+
+    pub fn get_all_blocks(&self) -> Option<Vec<u64>> {
+        self.blocks.as_ref().map(|blocks| blocks.values().flatten().cloned().collect())
+    }
+
+    pub fn get_blocks_by_epoch(&self, epoch: u64) -> Option<Vec<u64>> {
+        self.blocks
+            .as_ref()
+            .and_then(|blocks| blocks.get(&epoch).map(|blocks| blocks.iter().cloned().collect()))
+    }
+
+    pub fn add_block(&mut self, epoch: u64, block_number: u64) -> Option<()> {
+        self.blocks.as_mut().and_then(|blocks| {
+            blocks.entry(epoch).or_insert_with(Vector::new).push_back(block_number);
+            Some(())
+        })
+    }
 }
diff --git a/modules/spo_state/src/spo_state.rs b/modules/spo_state/src/spo_state.rs
@@ -634,12 +634,19 @@ impl SPOState {
                         PoolsStateQueryResponse::PoolTotalBlocksMinted(state.get_total_blocks_minted_by_pool(&pool_id))
                     }
 
-                    PoolsStateQuery::GetPoolBlockHashes { pool_id } => {
-                        if state.is_block_hashes_enabled() {
-                            PoolsStateQueryResponse::PoolBlockHashes(state.get_pool_block_hashes(pool_id).unwrap_or_default())
-                        } else {
-                            PoolsStateQueryResponse::Error("Block hashes are not enabled".into())
-                        }
+                    PoolsStateQuery::GetBlocksByPool { pool_id } => {
+                        state
+                            .is_historical_blocks_enabled()
+                            .then(|| PoolsStateQueryResponse::BlocksByPool(state.get_blocks_by_pool(pool_id).unwrap_or_default()))
+                            .unwrap_or(PoolsStateQueryResponse::Error("Blocks are not enabled".into()))
+                    }
+
+                    PoolsStateQuery::GetBlocksByPoolAndEpoch { pool_id, epoch } => {
+                        state
+                        .is_historical_blocks_enabled()
+                        .then(|| PoolsStateQueryResponse::BlocksByPoolAndEpoch(state.get_blocks_by_pool_and_epoch(pool_id, *epoch)
+                        .unwrap_or_default()))
+                        .unwrap_or(PoolsStateQueryResponse::Error("Blocks are not enabled".into()))
                     }
 
                     PoolsStateQuery::GetPoolUpdates { pool_id } => {
diff --git a/modules/spo_state/src/state.rs b/modules/spo_state/src/state.rs
diff --git a/modules/spo_state/src/store_config.rs b/modules/spo_state/src/store_config.rs
diff --git a/modules/spo_state/src/test_utils.rs b/modules/spo_state/src/test_utils.rs
diff --git a/processes/omnibus/omnibus.toml b/processes/omnibus/omnibus.toml
