fix(block_producer)

re: #124

Author: binier

node/native/src/block_producer/mod.rs

@@ -1,7 +1,7 @@
 mod vrf_evaluator;
 
 use mina_signer::Keypair;
-use node::{core::channels::mpsc, block_producer::vrf_evaluator::VrfEvaluatorInput};
+use node::{block_producer::vrf_evaluator::VrfEvaluatorInput, core::channels::mpsc};
 
 use crate::NodeService;
 

node/native/src/block_producer/vrf_evaluator.rs

@@ -1,5 +1,8 @@
 use mina_signer::Keypair;
-use node::{block_producer::{BlockProducerEvent, vrf_evaluator::VrfEvaluatorInput}, event_source::Event};
+use node::{
+    block_producer::{vrf_evaluator::VrfEvaluatorInput, BlockProducerEvent},
+    event_source::Event,
+};
 use openmina_core::channels::mpsc::{UnboundedReceiver, UnboundedSender};
 use vrf::{VrfEvaluationInput, VrfEvaluationOutput};
 

node/src/block_producer/block_producer_actions.rs

@@ -66,32 +66,10 @@ impl redux::EnablingCondition<crate::State> for BlockProducerWonSlotAction {
             let Some(best_tip) = state.transition_frontier.best_tip() else {
                 return false;
             };
-            let is_won_slot_old = |won_slot: &BlockProducerWonSlot| {
-                best_tip.global_slot() > won_slot.global_slot_since_genesis.as_u32()
-            };
 
-            state.time() < self.won_slot.next_slot_time()
-                && is_won_slot_old(&self.won_slot)
-                && this
-                    .current
-                    .won_slot()
-                    .filter(|won_slot| !is_won_slot_old(won_slot))
-                    .map_or(true, |won_slot| {
-                        match self
-                            .won_slot
-                            .global_slot_since_genesis
-                            .as_u32()
-                            .cmp(&won_slot.global_slot_since_genesis.as_u32())
-                        {
-                            // old won_slot is further in the future, pick new one instead.
-                            Ordering::Less => true,
-                            Ordering::Equal => &self.won_slot > won_slot,
-                            // new won_slot is further in the future. Ignore it.
-                            Ordering::Greater => false,
-                        }
-                    })
-            // TODO(binier): do we need to check if staking epoch ledger for an
-            // existing won slot is still current best tip's staking epoch ledger.
+            this.current.won_slot_should_search()
+                && state.time() < self.won_slot.next_slot_time()
+                && self.won_slot.global_slot_since_genesis.as_u32() >= best_tip.global_slot()
         })
     }
 }
@@ -113,7 +91,7 @@ pub struct BlockProducerWonSlotProduceInitAction {}
 impl redux::EnablingCondition<crate::State> for BlockProducerWonSlotProduceInitAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
         state.block_producer.with(false, |this| {
-            this.current.won_slot().is_some() && !this.current.won_slot_should_wait(state.time())
+            this.current.won_slot_should_produce(state.time())
         })
     }
 }
@@ -199,10 +177,7 @@ pub struct BlockProducerBlockInjectAction {}
 impl redux::EnablingCondition<crate::State> for BlockProducerBlockInjectAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
         state.block_producer.with(false, |this| {
-            matches!(
-                this.current,
-                BlockProducerCurrentState::BlockUnprovenBuilt { .. }
-            )
+            matches!(this.current, BlockProducerCurrentState::Produced { .. })
         })
     }
 }
@@ -213,10 +188,7 @@ pub struct BlockProducerBlockInjectedAction {}
 impl redux::EnablingCondition<crate::State> for BlockProducerBlockInjectedAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
         state.block_producer.with(false, |this| {
-            matches!(
-                this.current,
-                BlockProducerCurrentState::BlockUnprovenBuilt { .. }
-            )
+            matches!(this.current, BlockProducerCurrentState::Produced { .. })
         })
     }
 }
@@ -228,7 +200,11 @@ pub struct BlockProducerWonSlotDiscardAction {
 
 impl redux::EnablingCondition<crate::State> for BlockProducerWonSlotDiscardAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
-        todo!("validate reason")
+        let reason = state.block_producer.with(None, |bp| {
+            let best_tip = state.transition_frontier.best_tip()?;
+            bp.current.won_slot_should_discard(best_tip)
+        });
+        Some(&self.reason) == reason.as_ref()
     }
 }
 

node/src/block_producer/block_producer_effects.rs

@@ -14,8 +14,8 @@ use super::{
     BlockProducerStagedLedgerDiffCreatePendingAction,
     BlockProducerStagedLedgerDiffCreateSuccessAction, BlockProducerWonSlot,
     BlockProducerWonSlotAction, BlockProducerWonSlotDiscardAction,
-    BlockProducerWonSlotProduceInitAction, BlockProducerWonSlotSearchAction,
-    BlockProducerWonSlotWaitAction,
+    BlockProducerWonSlotDiscardReason, BlockProducerWonSlotProduceInitAction,
+    BlockProducerWonSlotSearchAction, BlockProducerWonSlotWaitAction,
 };
 
 pub fn block_producer_effects<S: crate::Service>(
@@ -92,29 +92,31 @@ impl BlockProducerBestTipUpdateAction {
     pub fn effects<S: redux::Service>(self, _: &ActionMeta, store: &mut Store<S>) {
         let protocol_state = &self.best_tip.block.header.protocol_state.body;
 
-        let current_epoch = store
-            .state()
-            .block_producer
-            .with(None, |this| this.vrf_evaluator.current_epoch);
+        let vrf_evaluator_epoch = store.state().block_producer.with(None, |this| {
+            Some(&this.vrf_evaluator.current_epoch_data.ledger)
+        });
 
-        // on new run when no current_epoch is set
-        if current_epoch.is_none() {
+        if vrf_evaluator_epoch.cloned()
+            != Some(
+                protocol_state
+                    .consensus_state
+                    .staking_epoch_data
+                    .ledger
+                    .hash
+                    .to_string(),
+            )
+        {
             store.dispatch(BlockProducerVrfEvaluatorEpochDataUpdateAction {
                 new_epoch_number: protocol_state.consensus_state.epoch_count.as_u32(),
                 epoch_data: protocol_state.consensus_state.staking_epoch_data.clone(),
                 next_epoch_data: protocol_state.consensus_state.next_epoch_data.clone(),
             });
         }
 
-        // on epoch change
-        if let Some(current_epoch) = current_epoch {
-            if current_epoch != protocol_state.consensus_state.epoch_count.as_u32() {
-                store.dispatch(BlockProducerVrfEvaluatorEpochDataUpdateAction {
-                    new_epoch_number: protocol_state.consensus_state.epoch_count.as_u32(),
-                    epoch_data: protocol_state.consensus_state.staking_epoch_data.clone(),
-                    next_epoch_data: protocol_state.consensus_state.next_epoch_data.clone(),
-                });
-            }
+        if let Some(reason) = store.state().block_producer.with(None, |bp| {
+            bp.current.won_slot_should_discard(&self.best_tip)
+        }) {
+            store.dispatch(BlockProducerWonSlotDiscardAction { reason });
         }
     }
 }

node/src/block_producer/block_producer_reducer.rs

@@ -171,11 +171,8 @@ impl BlockProducerEnabled {
                 let global_slot_since_genesis = won_slot.global_slot_since_genesis.clone();
                 let (pred_epoch, _) = to_epoch_and_slot(&pred_global_slot);
                 let (next_epoch, next_slot) = to_epoch_and_slot(&curr_global_slot);
-                let has_ancestor_in_same_checkpoint_window = in_same_checkpoint_window(
-                    &pred_global_slot,
-                    &curr_global_slot,
-                    pred_block.constants(),
-                );
+                let has_ancestor_in_same_checkpoint_window =
+                    in_same_checkpoint_window(&pred_global_slot, &curr_global_slot);
 
                 let block_stake_winner = won_slot.delegator.0.clone();
                 let vrf_truncated_output = won_slot.vrf_output.clone();
@@ -440,7 +437,7 @@ impl BlockProducerEnabled {
                         time: meta.time(),
                         won_slot: won_slot.clone(),
                         chain: std::mem::take(chain),
-                        block: block.clone(),
+                        block: dbg!(block).clone(),
                     };
                 }
             }
@@ -453,7 +450,7 @@ impl BlockProducerEnabled {
                     ..
                 } = &mut self.current
                 {
-                    self.current = BlockProducerCurrentState::Produced {
+                    self.current = BlockProducerCurrentState::Injected {
                         time: meta.time(),
                         won_slot: won_slot.clone(),
                         chain: std::mem::take(chain),
@@ -517,15 +514,11 @@ fn in_seed_update_range(
 fn in_same_checkpoint_window(
     slot1: &ConsensusGlobalSlotStableV1,
     slot2: &ConsensusGlobalSlotStableV1,
-    constants: &MinaBaseProtocolConstantsCheckedValueStableV1,
 ) -> bool {
-    checkpoint_window(slot1, constants) == checkpoint_window(slot2, constants)
+    checkpoint_window(slot1) == checkpoint_window(slot2)
 }
 
-fn checkpoint_window(
-    slot: &ConsensusGlobalSlotStableV1,
-    constants: &MinaBaseProtocolConstantsCheckedValueStableV1,
-) -> u32 {
+fn checkpoint_window(slot: &ConsensusGlobalSlotStableV1) -> u32 {
     slot.slot_number.as_u32() / checkpoint_window_size_in_slots()
 }
 
@@ -537,7 +530,7 @@ fn checkpoint_window_size_in_slots() -> u32 {
     let one_year_ms = days_to_ms(365);
     let slots_per_year = one_year_ms / CONSTRAINT_CONSTANTS.block_window_duration_ms;
     let size_in_slots = slots_per_year / 12;
-    assert_eq!(size_in_slots % 12, 0);
+    assert_eq!(slots_per_year % 12, 0);
     size_in_slots as u32
 }
 

node/src/block_producer/block_producer_state.rs

@@ -2,7 +2,7 @@ use mina_p2p_messages::v2::{
     ConsensusBodyReferenceStableV1, LedgerProofProdStableV2, MinaBaseStagedLedgerHashStableV1,
     NonZeroCurvePoint, StagedLedgerDiffDiffStableV2,
 };
-use openmina_core::block::ArcBlockWithHash;
+use openmina_core::{block::ArcBlockWithHash, consensus::consensus_take};
 use serde::{Deserialize, Serialize};
 
 use super::{
@@ -85,8 +85,12 @@ pub enum BlockProducerCurrentState {
     },
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
-pub enum BlockProducerWonSlotDiscardReason {}
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+pub enum BlockProducerWonSlotDiscardReason {
+    BestTipStakingLedgerDifferent,
+    BestTipGlobalSlotHigher,
+    BestTipSuperior,
+}
 
 impl BlockProducerState {
     pub fn new(now: redux::Timestamp, config: Option<BlockProducerConfig>) -> Self {
@@ -197,11 +201,46 @@ impl BlockProducerCurrentState {
 
     pub fn won_slot_should_wait(&self, now: redux::Timestamp) -> bool {
         match self {
-            Self::WonSlot { won_slot, .. } => won_slot.slot_time < now,
+            Self::WonSlot { won_slot, .. } => now < won_slot.slot_time,
+            _ => false,
+        }
+    }
+
+    pub fn won_slot_should_produce(&self, now: redux::Timestamp) -> bool {
+        match self {
+            Self::WonSlot { won_slot, .. } | Self::WonSlotWait { won_slot, .. } => {
+                now >= won_slot.slot_time
+            }
             _ => false,
         }
     }
 
+    pub fn won_slot_should_discard(
+        &self,
+        best_tip: &ArcBlockWithHash,
+    ) -> Option<BlockProducerWonSlotDiscardReason> {
+        let won_slot = self.won_slot()?;
+        if won_slot.global_slot_since_genesis.as_u32() < best_tip.global_slot() {
+            return Some(BlockProducerWonSlotDiscardReason::BestTipGlobalSlotHigher);
+        }
+        // TODO(binier): check if staking ledger changed
+
+        if won_slot < best_tip
+            || self.produced_block().map_or(false, |block| {
+                !consensus_take(
+                    best_tip.consensus_state(),
+                    block.consensus_state(),
+                    best_tip.hash(),
+                    block.hash(),
+                )
+            })
+        {
+            return Some(BlockProducerWonSlotDiscardReason::BestTipSuperior);
+        }
+
+        None
+    }
+
     pub fn is_producing(&self) -> bool {
         match self {
             Self::Idle { .. }

node/src/transition_frontier/sync/transition_frontier_sync_actions.rs

@@ -1,5 +1,6 @@
 use mina_p2p_messages::v2::StateHash;
 use openmina_core::block::ArcBlockWithHash;
+use openmina_core::consensus::consensus_take;
 use serde::{Deserialize, Serialize};
 
 use crate::p2p::channels::rpc::P2pRpcId;
@@ -91,10 +92,15 @@ impl redux::EnablingCondition<crate::State> for TransitionFrontierSyncBestTipUpd
                 .sync
                 .best_tip()
                 .map_or(true, |tip| self.best_tip.hash != tip.hash)
+            // TODO(binier): TMP. we shouldn't need to check consensus here.
             && state
-                .consensus
+                .transition_frontier
+                .sync
                 .best_tip()
-                .map_or(true, |tip| &self.best_tip.hash == tip.hash)
+                .or(state.transition_frontier.best_tip())
+                .map_or(false, |tip| {
+                    consensus_take(tip.consensus_state(), self.best_tip.consensus_state(), tip.hash(), self.best_tip.hash())
+                })
             // Don't sync to best tip if we are in the middle of producing
             // a block unless that best tip candidate is better consensus-wise
             // than the one that we are producing.