refactor: check blocks length on epochs blocks endpoint, use single block in block message configguration key

golddydev · golddydev · commit b356cd766171 · 2025-11-19T14:36:36.000+01:00
diff --git a/modules/block_kes_validator/src/block_kes_validator.rs b/modules/block_kes_validator/src/block_kes_validator.rs
@@ -26,8 +26,8 @@ const DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC: (&str, &str) = (
     "bootstrapped-subscribe-topic",
     "cardano.sequence.bootstrapped",
 );
-const DEFAULT_BLOCKS_SUBSCRIBE_TOPIC: (&str, &str) =
-    ("blocks-subscribe-topic", "cardano.block.proposed");
+const DEFAULT_BLOCK_SUBSCRIBE_TOPIC: (&str, &str) =
+    ("block-subscribe-topic", "cardano.block.proposed");
 const DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC: (&str, &str) = (
     "protocol-parameters-subscribe-topic",
     "cardano.protocol.parameters",
@@ -50,7 +50,7 @@ impl BlockKesValidator {
         history: Arc<Mutex<StateHistory<State>>>,
         kes_validation_publisher: KesValidationPublisher,
         mut bootstrapped_subscription: Box<dyn Subscription<Message>>,
-        mut blocks_subscription: Box<dyn Subscription<Message>>,
+        mut block_subscription: Box<dyn Subscription<Message>>,
         mut protocol_parameters_subscription: Box<dyn Subscription<Message>>,
         mut spo_state_subscription: Box<dyn Subscription<Message>>,
     ) -> Result<()> {
@@ -70,7 +70,7 @@ impl BlockKesValidator {
             let mut state = history.lock().await.get_or_init_with(State::new);
             let mut current_block: Option<BlockInfo> = None;
 
-            let (_, message) = blocks_subscription.read().await?;
+            let (_, message) = block_subscription.read().await?;
             match message.as_ref() {
                 Message::Cardano((block_info, CardanoMessage::BlockAvailable(block_msg))) => {
                     // handle rollback here
@@ -160,10 +160,10 @@ impl BlockKesValidator {
             .unwrap_or(DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC.1.to_string());
         info!("Creating subscriber for bootstrapped on '{bootstrapped_subscribe_topic}'");
 
-        let blocks_subscribe_topic = config
-            .get_string(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.0)
-            .unwrap_or(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.1.to_string());
-        info!("Creating blocks subscription on '{blocks_subscribe_topic}'");
+        let block_subscribe_topic = config
+            .get_string(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.0)
+            .unwrap_or(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.1.to_string());
+        info!("Creating block subscription on '{block_subscribe_topic}'");
 
         let protocol_parameters_subscribe_topic = config
             .get_string(DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC.0)
@@ -181,7 +181,7 @@ impl BlockKesValidator {
 
         // Subscribers
         let bootstrapped_subscription = context.subscribe(&bootstrapped_subscribe_topic).await?;
-        let blocks_subscription = context.subscribe(&blocks_subscribe_topic).await?;
+        let block_subscription = context.subscribe(&block_subscribe_topic).await?;
         let protocol_parameters_subscription =
             context.subscribe(&protocol_parameters_subscribe_topic).await?;
         let spo_state_subscription = context.subscribe(&spo_state_subscribe_topic).await?;
@@ -198,7 +198,7 @@ impl BlockKesValidator {
                 history,
                 kes_validation_publisher,
                 bootstrapped_subscription,
-                blocks_subscription,
+                block_subscription,
                 protocol_parameters_subscription,
                 spo_state_subscription,
             )
diff --git a/modules/block_vrf_validator/src/block_vrf_validator.rs b/modules/block_vrf_validator/src/block_vrf_validator.rs
@@ -27,8 +27,8 @@ const DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC: (&str, &str) = (
     "bootstrapped-subscribe-topic",
     "cardano.sequence.bootstrapped",
 );
-const DEFAULT_BLOCKS_SUBSCRIBE_TOPIC: (&str, &str) =
-    ("blocks-subscribe-topic", "cardano.block.proposed");
+const DEFAULT_BLOCK_SUBSCRIBE_TOPIC: (&str, &str) =
+    ("block-subscribe-topic", "cardano.block.proposed");
 const DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC: (&str, &str) = (
     "protocol-parameters-subscribe-topic",
     "cardano.protocol.parameters",
@@ -55,7 +55,7 @@ impl BlockVrfValidator {
         history: Arc<Mutex<StateHistory<State>>>,
         mut vrf_validation_publisher: VrfValidationPublisher,
         mut bootstrapped_subscription: Box<dyn Subscription<Message>>,
-        mut blocks_subscription: Box<dyn Subscription<Message>>,
+        mut block_subscription: Box<dyn Subscription<Message>>,
         mut protocol_parameters_subscription: Box<dyn Subscription<Message>>,
         mut epoch_activity_subscription: Box<dyn Subscription<Message>>,
         mut spo_state_subscription: Box<dyn Subscription<Message>>,
@@ -77,7 +77,7 @@ impl BlockVrfValidator {
             let mut state = history.lock().await.get_or_init_with(State::new);
             let mut current_block: Option<BlockInfo> = None;
 
-            let (_, message) = blocks_subscription.read().await?;
+            let (_, message) = block_subscription.read().await?;
             match message.as_ref() {
                 Message::Cardano((block_info, CardanoMessage::BlockAvailable(block_msg))) => {
                     // handle rollback here
@@ -190,10 +190,10 @@ impl BlockVrfValidator {
             .unwrap_or(DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC.1.to_string());
         info!("Creating subscriber for protocol parameters on '{protocol_parameters_subscribe_topic}'");
 
-        let blocks_subscribe_topic = config
-            .get_string(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.0)
-            .unwrap_or(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.1.to_string());
-        info!("Creating blocks subscription on '{blocks_subscribe_topic}'");
+        let block_subscribe_topic = config
+            .get_string(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.0)
+            .unwrap_or(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.1.to_string());
+        info!("Creating block subscription on '{block_subscribe_topic}'");
 
         let epoch_activity_subscribe_topic = config
             .get_string(DEFAULT_EPOCH_ACTIVITY_SUBSCRIBE_TOPIC.0)
@@ -218,7 +218,7 @@ impl BlockVrfValidator {
         let bootstrapped_subscription = context.subscribe(&bootstrapped_subscribe_topic).await?;
         let protocol_parameters_subscription =
             context.subscribe(&protocol_parameters_subscribe_topic).await?;
-        let blocks_subscription = context.subscribe(&blocks_subscribe_topic).await?;
+        let block_subscription = context.subscribe(&block_subscribe_topic).await?;
         let epoch_activity_subscription =
             context.subscribe(&epoch_activity_subscribe_topic).await?;
         let spo_state_subscription = context.subscribe(&spo_state_subscribe_topic).await?;
@@ -236,7 +236,7 @@ impl BlockVrfValidator {
                 history,
                 vrf_validation_publisher,
                 bootstrapped_subscription,
-                blocks_subscription,
+                block_subscription,
                 protocol_parameters_subscription,
                 epoch_activity_subscription,
                 spo_state_subscription,
diff --git a/modules/chain_store/src/stores/fjall.rs b/modules/chain_store/src/stores/fjall.rs
@@ -193,6 +193,13 @@ impl FjallBlockStore {
     }
 
     fn get_by_number_range(&self, min_number: u64, max_number: u64) -> Result<Vec<Block>> {
+        if max_number < min_number {
+            return Err(anyhow::anyhow!(
+                "Invalid number range min={min_number}, max={max_number}"
+            ));
+        }
+        let expected_count = max_number - min_number + 1;
+
         let min_number_bytes = min_number.to_be_bytes();
         let max_number_bytes = max_number.to_be_bytes();
         let mut blocks = vec![];
@@ -202,6 +209,12 @@ impl FjallBlockStore {
                 blocks.push(block);
             }
         }
+        if blocks.len() as u64 != expected_count {
+            return Err(anyhow::anyhow!(
+                "Expected {expected_count} blocks, got {}",
+                blocks.len()
+            ));
+        }
         Ok(blocks)
     }
 
diff --git a/modules/epochs_state/src/epochs_state.rs b/modules/epochs_state/src/epochs_state.rs
@@ -32,8 +32,8 @@ const DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC: (&str, &str) = (
     "bootstrapped-subscribe-topic",
     "cardano.sequence.bootstrapped",
 );
-const DEFAULT_BLOCKS_SUBSCRIBE_TOPIC: (&str, &str) =
-    ("blocks-subscribe-topic", "cardano.block.proposed");
+const DEFAULT_BLOCK_SUBSCRIBE_TOPIC: (&str, &str) =
+    ("block-subscribe-topic", "cardano.block.proposed");
 const DEFAULT_BLOCK_TXS_SUBSCRIBE_TOPIC: (&str, &str) =
     ("block-txs-subscribe-topic", "cardano.block.txs");
 const DEFAULT_PROTOCOL_PARAMETERS_SUBSCRIBE_TOPIC: (&str, &str) = (
@@ -58,7 +58,7 @@ impl EpochsState {
         history: Arc<Mutex<StateHistory<State>>>,
         epochs_history: EpochsHistoryState,
         mut bootstrapped_subscription: Box<dyn Subscription<Message>>,
-        mut blocks_subscription: Box<dyn Subscription<Message>>,
+        mut block_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,
@@ -80,11 +80,11 @@ impl EpochsState {
             let mut current_block: Option<BlockInfo> = None;
 
             // Read both topics in parallel
-            let blocks_message_f = blocks_subscription.read();
+            let block_message_f = block_subscription.read();
             let block_txs_message_f = block_txs_subscription.read();
 
             // Handle blocks first
-            let (_, message) = blocks_message_f.await?;
+            let (_, message) = block_message_f.await?;
             match message.as_ref() {
                 Message::Cardano((block_info, CardanoMessage::BlockAvailable(block_msg))) => {
                     // handle rollback here
@@ -188,10 +188,10 @@ impl EpochsState {
             .unwrap_or(DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC.1.to_string());
         info!("Creating subscriber for bootstrapped on '{bootstrapped_subscribe_topic}'");
 
-        let blocks_subscribe_topic = config
-            .get_string(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.0)
-            .unwrap_or(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.1.to_string());
-        info!("Creating subscriber for blocks on '{blocks_subscribe_topic}'");
+        let block_subscribe_topic = config
+            .get_string(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.0)
+            .unwrap_or(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.1.to_string());
+        info!("Creating subscriber for blocks on '{block_subscribe_topic}'");
 
         let block_txs_subscribe_topic = config
             .get_string(DEFAULT_BLOCK_TXS_SUBSCRIBE_TOPIC.0)
@@ -231,7 +231,7 @@ impl EpochsState {
 
         // Subscribe
         let bootstrapped_subscription = context.subscribe(&bootstrapped_subscribe_topic).await?;
-        let blocks_subscription = context.subscribe(&blocks_subscribe_topic).await?;
+        let block_subscription = context.subscribe(&block_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?;
@@ -342,7 +342,7 @@ impl EpochsState {
                 history,
                 epochs_history,
                 bootstrapped_subscription,
-                blocks_subscription,
+                block_subscription,
                 block_txs_subscription,
                 protocol_parameters_subscription,
                 epoch_activity_publisher,
diff --git a/modules/epochs_state/src/state.rs b/modules/epochs_state/src/state.rs
@@ -69,9 +69,13 @@ 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,
+            // NOTE:
+            // First block height is 1
+            // only because we don't handle EBB for now
+            // so by default, we counter epoch 0's EBB
+            first_block_height: 1,
+            last_block_time: genesis.byron_timestamp,
+            last_block_height: 1,
             blocks_minted: HashMap::new(),
             epoch_blocks: 0,
             epoch_txs: 0,
diff --git a/modules/mithril_snapshot_fetcher/src/mithril_snapshot_fetcher.rs b/modules/mithril_snapshot_fetcher/src/mithril_snapshot_fetcher.rs
@@ -30,9 +30,13 @@ use tracing::{debug, error, info, info_span, Instrument};
 mod pause;
 use pause::PauseType;
 
-const DEFAULT_STARTUP_TOPIC: &str = "cardano.sequence.bootstrapped";
-const DEFAULT_BLOCK_TOPIC: &str = "cardano.block.available";
-const DEFAULT_COMPLETION_TOPIC: &str = "cardano.snapshot.complete";
+const DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC: (&str, &str) = (
+    "bootstrapped-subscribe-topic",
+    "cardano.sequence.bootstrapped",
+);
+const DEFAULT_BLOCK_PUBLISH_TOPIC: (&str, &str) =
+    ("block-publish-topic", "cardano.block.available");
+const DEFAULT_COMPLETION_TOPIC: (&str, &str) = ("completion-topic", "cardano.snapshot.complete");
 
 const DEFAULT_AGGREGATOR_URL: &str =
     "https://aggregator.release-mainnet.api.mithril.network/aggregator";
@@ -237,12 +241,14 @@ impl MithrilSnapshotFetcher {
         config: Arc<Config>,
         genesis: GenesisValues,
     ) -> Result<()> {
-        let block_topic =
-            config.get_string("block-topic").unwrap_or(DEFAULT_BLOCK_TOPIC.to_string());
-        info!("Publishing blocks on '{block_topic}'");
-
-        let completion_topic =
-            config.get_string("completion-topic").unwrap_or(DEFAULT_COMPLETION_TOPIC.to_string());
+        let block_publish_topic = config
+            .get_string(DEFAULT_BLOCK_PUBLISH_TOPIC.0)
+            .unwrap_or(DEFAULT_BLOCK_PUBLISH_TOPIC.1.to_string());
+        info!("Publishing blocks on '{block_publish_topic}'");
+
+        let completion_topic = config
+            .get_string(DEFAULT_COMPLETION_TOPIC.0)
+            .unwrap_or(DEFAULT_COMPLETION_TOPIC.1.to_string());
         info!("Publishing completion on '{completion_topic}'");
 
         let directory = config.get_string("directory").unwrap_or(DEFAULT_DIRECTORY.to_string());
@@ -356,7 +362,7 @@ impl MithrilSnapshotFetcher {
 
                         context
                             .message_bus
-                            .publish(&block_topic, Arc::new(message_enum))
+                            .publish(&block_publish_topic, Arc::new(message_enum))
                             .await
                             .unwrap_or_else(|e| error!("Failed to publish block message: {e}"));
 
@@ -389,21 +395,23 @@ impl MithrilSnapshotFetcher {
 
     /// Main init function
     pub async fn init(&self, context: Arc<Context<Message>>, config: Arc<Config>) -> Result<()> {
-        let startup_topic =
-            config.get_string("startup-topic").unwrap_or(DEFAULT_STARTUP_TOPIC.to_string());
-        info!("Creating startup subscriber on '{startup_topic}'");
+        let bootstrapped_subscribe_topic = config
+            .get_string(DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC.0)
+            .unwrap_or(DEFAULT_BOOTSTRAPPED_SUBSCRIBE_TOPIC.1.to_string());
+        info!("Creating subscriber for bootstrapped on '{bootstrapped_subscribe_topic}'");
 
-        let mut subscription = context.subscribe(&startup_topic).await?;
+        let mut bootstrapped_subscription =
+            context.subscribe(&bootstrapped_subscribe_topic).await?;
         context.clone().run(async move {
-            let Ok((_, startup_message)) = subscription.read().await else {
+            let Ok((_, bootstrapped_message)) = bootstrapped_subscription.read().await else {
                 return;
             };
-            info!("Received startup message");
-            let genesis = match startup_message.as_ref() {
+            info!("Received bootstrapped message");
+            let genesis = match bootstrapped_message.as_ref() {
                 Message::Cardano((_, CardanoMessage::GenesisComplete(complete))) => {
                     complete.values.clone()
                 }
-                x => panic!("unexpected startup message: {x:?}"),
+                x => panic!("unexpected bootstrapped message: {x:?}"),
             };
 
             let mut delay = 1;
diff --git a/modules/spo_state/src/spo_state.rs b/modules/spo_state/src/spo_state.rs
@@ -47,8 +47,8 @@ const DEFAULT_WITHDRAWALS_SUBSCRIBE_TOPIC: (&str, &str) =
     ("withdrawals-subscribe-topic", "cardano.withdrawals");
 const DEFAULT_GOVERNANCE_SUBSCRIBE_TOPIC: (&str, &str) =
     ("governance-subscribe-topic", "cardano.governance");
-const DEFAULT_BLOCKS_SUBSCRIBE_TOPIC: (&str, &str) =
-    ("blocks-subscribe-topic", "cardano.block.proposed");
+const DEFAULT_BLOCK_SUBSCRIBE_TOPIC: (&str, &str) =
+    ("block-subscribe-topic", "cardano.block.proposed");
 const DEFAULT_EPOCH_ACTIVITY_SUBSCRIBE_TOPIC: (&str, &str) =
     ("epoch-activity-subscribe-topic", "cardano.epoch.activity");
 const DEFAULT_SPDD_SUBSCRIBE_TOPIC: (&str, &str) =
@@ -87,7 +87,7 @@ impl SPOState {
         store_config: &StoreConfig,
         // subscribers
         mut certificates_subscription: Box<dyn Subscription<Message>>,
-        mut blocks_subscription: Box<dyn Subscription<Message>>,
+        mut block_subscription: Box<dyn Subscription<Message>>,
         mut withdrawals_subscription: Option<Box<dyn Subscription<Message>>>,
         mut governance_subscription: Option<Box<dyn Subscription<Message>>>,
         mut epoch_activity_subscription: Box<dyn Subscription<Message>>,
@@ -113,7 +113,7 @@ impl SPOState {
 
             // read per-block topics in parallel
             let certs_message_f = certificates_subscription.read();
-            let blocks_message_f = blocks_subscription.read();
+            let block_message_f = block_subscription.read();
             let withdrawals_message_f = withdrawals_subscription.as_mut().map(|s| s.read());
             let governance_message_f = governance_subscription.as_mut().map(|s| s.read());
             let stake_deltas_message_f = stake_deltas_subscription.as_mut().map(|s| s.read());
@@ -140,7 +140,7 @@ impl SPOState {
 
             // handle blocks (handle_mint) before handle_tx_certs
             // in case of epoch boundary
-            let (_, block_message) = blocks_message_f.await?;
+            let (_, block_message) = block_message_f.await?;
             match block_message.as_ref() {
                 Message::Cardano((block_info, CardanoMessage::BlockAvailable(block_msg))) => {
                     let span =
@@ -438,10 +438,10 @@ impl SPOState {
             .unwrap_or(DEFAULT_SPO_REWARDS_SUBSCRIBE_TOPIC.1.to_string());
         info!("Creating SPO rewards subscriber on '{spo_rewards_subscribe_topic}'");
 
-        let blocks_subscribe_topic = config
-            .get_string(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.0)
-            .unwrap_or(DEFAULT_BLOCKS_SUBSCRIBE_TOPIC.1.to_string());
-        info!("Creating blocks subscriber on '{blocks_subscribe_topic}'");
+        let block_subscribe_topic = config
+            .get_string(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.0)
+            .unwrap_or(DEFAULT_BLOCK_SUBSCRIBE_TOPIC.1.to_string());
+        info!("Creating block subscriber on '{block_subscribe_topic}'");
 
         let stake_reward_deltas_subscribe_topic = config
             .get_string(DEFAULT_STAKE_REWARD_DELTAS_SUBSCRIBE_TOPIC.0)
@@ -765,7 +765,7 @@ impl SPOState {
 
         // Subscriptions
         let certificates_subscription = context.subscribe(&certificates_subscribe_topic).await?;
-        let blocks_subscription = context.subscribe(&blocks_subscribe_topic).await?;
+        let block_subscription = context.subscribe(&block_subscribe_topic).await?;
         let epoch_activity_subscription =
             context.subscribe(&epoch_activity_subscribe_topic).await?;
         let spdd_subscription = context.subscribe(&spdd_subscribe_topic).await?;
@@ -811,7 +811,7 @@ impl SPOState {
                 retired_pools_history,
                 &store_config,
                 certificates_subscription,
-                blocks_subscription,
+                block_subscription,
                 withdrawals_subscription,
                 governance_subscription,
                 epoch_activity_subscription,

Original file line number	Diff line number	Diff line change
`@@ -193,6 +193,13 @@ impl FjallBlockStore {`
`193`	`193`	`}`
`194`	`194`
`195`	`195`	`fn get_by_number_range(&self, min_number: u64, max_number: u64) -> Result<Vec<Block>> {`
	`196`	`+ if max_number < min_number {`
	`197`	`+ return Err(anyhow::anyhow!(`
	`198`	`+ "Invalid number range min={min_number}, max={max_number}"`
	`199`	`+ ));`
	`200`	`+ }`
	`201`	`+ let expected_count = max_number - min_number + 1;`
	`202`	`+`
`196`	`203`	`let min_number_bytes = min_number.to_be_bytes();`
`197`	`204`	`let max_number_bytes = max_number.to_be_bytes();`
`198`	`205`	`let mut blocks = vec![];`
`@@ -202,6 +209,12 @@ impl FjallBlockStore {`
`202`	`209`	`blocks.push(block);`
`203`	`210`	`}`
`204`	`211`	`}`
	`212`	`+ if blocks.len() as u64 != expected_count {`
	`213`	`+ return Err(anyhow::anyhow!(`
	`214`	`+ "Expected {expected_count} blocks, got {}",`
	`215`	`+ blocks.len()`
	`216`	`+ ));`
	`217`	`+ }`
`205`	`218`	`Ok(blocks)`
`206`	`219`	`}`
`207`	`220`