mina_node/
state.rs

1use std::{sync::Arc, time::Duration};
2
3use crate::p2p::P2pNetworkPubsubMessageCacheId;
4use malloc_size_of_derive::MallocSizeOf;
5use mina_core::{
6    block::prevalidate::{prevalidate_block, BlockPrevalidationError},
7    consensus::ConsensusTime,
8    transaction::{TransactionInfo, TransactionWithHash},
9};
10use mina_p2p_messages::v2;
11use rand::prelude::*;
12
13use crate::p2p::{
14    bootstrap::P2pNetworkKadBootstrapState,
15    channels::{
16        rpc::{P2pRpcId, P2pRpcRequest, P2pRpcResponse},
17        streaming_rpc::P2pStreamingRpcResponseFull,
18    },
19    connection::{outgoing::P2pConnectionOutgoingError, P2pConnectionResponse},
20    network::identify::P2pNetworkIdentifyState,
21    P2pCallbacks, P2pConfig, P2pNetworkSchedulerState, P2pPeerState, P2pPeerStatusReady, PeerId,
22};
23use mina_core::{
24    block::{ArcBlockWithHash, BlockWithHash},
25    consensus::ConsensusConstants,
26    constants::constraint_constants,
27    requests::RpcId,
28    snark::{Snark, SnarkInfo, SnarkJobCommitment},
29    ChainId,
30};
31use mina_snark::{
32    block_verify::SnarkBlockVerifyState, user_command_verify::SnarkUserCommandVerifyState,
33    work_verify::SnarkWorkVerifyState,
34};
35use redux::{ActionMeta, EnablingCondition, Timestamp};
36use serde::{Deserialize, Serialize};
37
38use crate::{
39    block_producer::vrf_evaluator::BlockProducerVrfEvaluatorState,
40    config::GlobalConfig,
41    external_snark_worker::{ExternalSnarkWorker, ExternalSnarkWorkers},
42    ledger::{read::LedgerReadState, write::LedgerWriteState},
43    p2p::callbacks::P2pCallbacksAction,
44    snark_pool::candidate::SnarkPoolCandidateAction,
45    transaction_pool::{
46        candidate::{TransactionPoolCandidateAction, TransactionPoolCandidatesState},
47        TransactionPoolState,
48    },
49    transition_frontier::{
50        candidate::TransitionFrontierCandidateAction,
51        genesis::TransitionFrontierGenesisState,
52        sync::{
53            ledger::{
54                snarked::TransitionFrontierSyncLedgerSnarkedState,
55                staged::TransitionFrontierSyncLedgerStagedState, TransitionFrontierSyncLedgerState,
56            },
57            TransitionFrontierSyncState,
58        },
59    },
60    ActionWithMeta, RpcAction, SnarkPoolAction,
61};
62pub use crate::{
63    block_producer::BlockProducerState,
64    ledger::LedgerState,
65    p2p::P2pState,
66    rpc::RpcState,
67    snark::SnarkState,
68    snark_pool::{candidate::SnarkPoolCandidatesState, SnarkPoolState},
69    transition_frontier::{candidate::TransitionFrontierCandidatesState, TransitionFrontierState},
70    watched_accounts::WatchedAccountsState,
71    Config,
72};
73
74#[derive(Serialize, Deserialize, Debug, Clone)]
75pub struct State {
76    pub config: GlobalConfig,
77
78    pub p2p: P2p,
79    pub ledger: LedgerState,
80    pub snark: SnarkState,
81    pub transition_frontier: TransitionFrontierState,
82    pub snark_pool: SnarkPoolState,
83    pub external_snark_worker: ExternalSnarkWorkers,
84    pub transaction_pool: TransactionPoolState,
85    pub block_producer: BlockProducerState,
86    pub rpc: RpcState,
87
88    pub watched_accounts: WatchedAccountsState,
89
90    // TODO(binier): include action kind in `last_action`.
91    last_action: ActionMeta,
92    applied_actions_count: u64,
93    start_time: Timestamp,
94}
95
96// Substate accessors that will be used in reducers
97use mina_core::{bug_condition, impl_substate_access, SubstateAccess};
98
99impl_substate_access!(State, SnarkState, snark);
100impl_substate_access!(State, SnarkBlockVerifyState, snark.block_verify);
101impl_substate_access!(State, SnarkWorkVerifyState, snark.work_verify);
102impl_substate_access!(
103    State,
104    SnarkUserCommandVerifyState,
105    snark.user_command_verify
106);
107impl_substate_access!(State, TransitionFrontierState, transition_frontier);
108impl_substate_access!(
109    State,
110    TransitionFrontierCandidatesState,
111    transition_frontier.candidates
112);
113impl_substate_access!(State, TransactionPoolState, transaction_pool);
114impl_substate_access!(
115    State,
116    TransactionPoolCandidatesState,
117    transaction_pool.candidates
118);
119impl_substate_access!(
120    State,
121    TransitionFrontierGenesisState,
122    transition_frontier.genesis
123);
124impl_substate_access!(State, TransitionFrontierSyncState, transition_frontier.sync);
125impl_substate_access!(State, SnarkPoolState, snark_pool);
126impl_substate_access!(State, SnarkPoolCandidatesState, snark_pool.candidates);
127impl_substate_access!(State, ExternalSnarkWorkers, external_snark_worker);
128impl_substate_access!(State, BlockProducerState, block_producer);
129impl_substate_access!(State, RpcState, rpc);
130impl_substate_access!(State, WatchedAccountsState, watched_accounts);
131impl_substate_access!(State, ExternalSnarkWorker, external_snark_worker.0);
132impl_substate_access!(State, LedgerState, ledger);
133impl_substate_access!(State, LedgerReadState, ledger.read);
134impl_substate_access!(State, LedgerWriteState, ledger.write);
135
136impl mina_core::SubstateAccess<P2pState> for State {
137    fn substate(&self) -> mina_core::SubstateResult<&P2pState> {
138        self.p2p
139            .ready()
140            .ok_or_else(|| "P2P state unavailable. P2P layer is not ready".to_owned())
141    }
142
143    fn substate_mut(&mut self) -> mina_core::SubstateResult<&mut P2pState> {
144        self.p2p
145            .ready_mut()
146            .ok_or_else(|| "P2P state unavailable. P2P layer is not ready".to_owned())
147    }
148}
149
150impl mina_core::SubstateAccess<TransitionFrontierSyncLedgerState> for State {
151    fn substate(&self) -> mina_core::SubstateResult<&TransitionFrontierSyncLedgerState> {
152        self.transition_frontier
153            .sync
154            .ledger()
155            .ok_or_else(|| "Ledger sync state unavailable".to_owned())
156    }
157
158    fn substate_mut(
159        &mut self,
160    ) -> mina_core::SubstateResult<&mut TransitionFrontierSyncLedgerState> {
161        self.transition_frontier
162            .sync
163            .ledger_mut()
164            .ok_or_else(|| "Ledger sync state unavailable".to_owned())
165    }
166}
167
168impl SubstateAccess<BlockProducerVrfEvaluatorState> for State {
169    fn substate(&self) -> mina_core::SubstateResult<&BlockProducerVrfEvaluatorState> {
170        self.block_producer
171            .as_ref()
172            .map(|state| &state.vrf_evaluator)
173            .ok_or_else(|| "Block producer VRF evaluator state unavailable".to_owned())
174    }
175
176    fn substate_mut(&mut self) -> mina_core::SubstateResult<&mut BlockProducerVrfEvaluatorState> {
177        self.block_producer
178            .as_mut()
179            .map(|state| &mut state.vrf_evaluator)
180            .ok_or_else(|| "Block producer VRF evaluator state unavailable".to_owned())
181    }
182}
183
184impl mina_core::SubstateAccess<TransitionFrontierSyncLedgerSnarkedState> for State {
185    fn substate(&self) -> mina_core::SubstateResult<&TransitionFrontierSyncLedgerSnarkedState> {
186        self.transition_frontier
187            .sync
188            .ledger()
189            .ok_or_else(|| {
190                "Snarked ledger state unavailable. Ledger sync state unavailable".to_owned()
191            })?
192            .snarked()
193            .ok_or_else(|| "Snarked ledger state unavailable".to_owned())
194    }
195
196    fn substate_mut(
197        &mut self,
198    ) -> mina_core::SubstateResult<&mut TransitionFrontierSyncLedgerSnarkedState> {
199        self.transition_frontier
200            .sync
201            .ledger_mut()
202            .ok_or_else(|| {
203                "Snarked ledger state unavailable. Ledger sync state unavailable".to_owned()
204            })?
205            .snarked_mut()
206            .ok_or_else(|| "Snarked ledger state unavailable".to_owned())
207    }
208}
209
210impl mina_core::SubstateAccess<TransitionFrontierSyncLedgerStagedState> for State {
211    fn substate(&self) -> mina_core::SubstateResult<&TransitionFrontierSyncLedgerStagedState> {
212        self.transition_frontier
213            .sync
214            .ledger()
215            .ok_or_else(|| {
216                "Staged ledger state unavailable. Ledger sync state unavailable".to_owned()
217            })?
218            .staged()
219            .ok_or_else(|| "Staged ledger state unavailable".to_owned())
220    }
221
222    fn substate_mut(
223        &mut self,
224    ) -> mina_core::SubstateResult<&mut TransitionFrontierSyncLedgerStagedState> {
225        self.transition_frontier
226            .sync
227            .ledger_mut()
228            .ok_or_else(|| {
229                "Staged ledger state unavailable. Ledger sync state unavailable".to_owned()
230            })?
231            .staged_mut()
232            .ok_or_else(|| "Staged ledger state unavailable".to_owned())
233    }
234}
235
236impl SubstateAccess<State> for State {
237    fn substate(&self) -> mina_core::SubstateResult<&State> {
238        Ok(self)
239    }
240
241    fn substate_mut(&mut self) -> mina_core::SubstateResult<&mut State> {
242        Ok(self)
243    }
244}
245
246macro_rules! impl_p2p_state_access {
247    ($state:ty, $substate_type:ty) => {
248        impl mina_core::SubstateAccess<$substate_type> for $state {
249            fn substate(&self) -> mina_core::SubstateResult<&$substate_type> {
250                let substate: &P2pState = self.substate()?;
251                substate.substate()
252            }
253
254            fn substate_mut(&mut self) -> mina_core::SubstateResult<&mut $substate_type> {
255                let substate: &mut P2pState = self.substate_mut()?;
256                substate.substate_mut()
257            }
258        }
259    };
260}
261
262impl_p2p_state_access!(State, P2pNetworkIdentifyState);
263impl_p2p_state_access!(State, crate::p2p::P2pNetworkState);
264impl_p2p_state_access!(State, P2pNetworkKadBootstrapState);
265impl_p2p_state_access!(State, crate::p2p::P2pNetworkKadState);
266impl_p2p_state_access!(State, P2pNetworkSchedulerState);
267impl_p2p_state_access!(State, crate::p2p::P2pLimits);
268impl_p2p_state_access!(State, crate::p2p::P2pNetworkPubsubState);
269impl_p2p_state_access!(State, crate::p2p::P2pConfig);
270
271impl crate::p2p::P2pStateTrait for State {}
272
273pub type Substate<'a, S> = mina_core::Substate<'a, crate::Action, State, S>;
274
275impl State {
276    pub fn new(config: Config, constants: &ConsensusConstants, now: Timestamp) -> Self {
277        Self {
278            p2p: P2p::Pending(config.p2p),
279            ledger: LedgerState::new(config.ledger),
280            snark_pool: SnarkPoolState::new(),
281            snark: SnarkState::new(config.snark),
282            transition_frontier: TransitionFrontierState::new(
283                config.transition_frontier,
284                config.archive.is_some(),
285            ),
286            external_snark_worker: ExternalSnarkWorkers::new(now),
287            block_producer: BlockProducerState::new(now, config.block_producer),
288            rpc: RpcState::new(),
289            transaction_pool: TransactionPoolState::new(config.tx_pool, constants),
290
291            watched_accounts: WatchedAccountsState::new(),
292
293            config: config.global,
294            last_action: ActionMeta::zero_custom(now),
295            start_time: now,
296            applied_actions_count: 0,
297        }
298    }
299
300    pub fn last_action(&self) -> &ActionMeta {
301        &self.last_action
302    }
303
304    /// Latest time observed by the state machine.
305    ///
306    /// Only updated when action is dispatched and reducer is executed.
307    #[inline(always)]
308    pub fn time(&self) -> Timestamp {
309        self.last_action.time()
310    }
311
312    pub fn pseudo_rng(&self) -> StdRng {
313        crate::core::pseudo_rng(self.time())
314    }
315
316    pub fn start_time(&self) -> Timestamp {
317        self.start_time
318    }
319
320    /// Must be called in the global reducer as the last thing only once
321    /// and only there!
322    pub fn action_applied(&mut self, action: &ActionWithMeta) {
323        self.last_action = action.meta().clone();
324        self.applied_actions_count = self.applied_actions_count.checked_add(1).expect("overflow");
325    }
326
327    pub fn genesis_block(&self) -> Option<ArcBlockWithHash> {
328        self.transition_frontier
329            .genesis
330            .block_with_real_or_dummy_proof()
331    }
332
333    fn cur_slot(&self, initial_slot: impl FnOnce(&ArcBlockWithHash) -> u32) -> Option<u32> {
334        let genesis = self.genesis_block()?;
335        let diff_ns = u64::from(self.time()).saturating_sub(u64::from(genesis.timestamp()));
336        let diff_ms = diff_ns / 1_000_000;
337        let slots = diff_ms
338            .checked_div(constraint_constants().block_window_duration_ms)
339            .expect("division by 0");
340        Some(
341            initial_slot(&genesis)
342                .checked_add(slots as u32)
343                .expect("overflow"),
344        )
345    }
346
347    /// Current global slot based on constants and current time.
348    ///
349    /// It's not equal to global slot of the best tip.
350    pub fn cur_global_slot(&self) -> Option<u32> {
351        self.cur_slot(|b| b.global_slot())
352    }
353
354    pub fn current_slot(&self) -> Option<u32> {
355        let slots_per_epoch = self.genesis_block()?.constants().slots_per_epoch.as_u32();
356        Some(
357            self.cur_global_slot()?
358                .checked_rem(slots_per_epoch)
359                .expect("division by 0"),
360        )
361    }
362
363    pub fn cur_global_slot_since_genesis(&self) -> Option<u32> {
364        self.cur_slot(|b| b.global_slot_since_genesis())
365    }
366
367    pub fn current_epoch(&self) -> Option<u32> {
368        let slots_per_epoch = self.genesis_block()?.constants().slots_per_epoch.as_u32();
369        Some(
370            self.cur_global_slot()?
371                .checked_div(slots_per_epoch)
372                .expect("division by 0"),
373        )
374    }
375
376    pub fn slot_time(&self, global_slot: u64) -> Option<(Timestamp, Timestamp)> {
377        let genesis_timestamp = self.genesis_block()?.genesis_timestamp();
378        println!("genesis_timestamp: {}", u64::from(genesis_timestamp));
379
380        let start_time = genesis_timestamp.checked_add(
381            global_slot
382                .checked_mul(constraint_constants().block_window_duration_ms)?
383                .checked_mul(1_000_000)?,
384        )?;
385        let end_time = start_time.checked_add(
386            constraint_constants()
387                .block_window_duration_ms
388                .checked_mul(1_000_000)?,
389        )?;
390
391        Some((start_time, end_time))
392    }
393
394    pub fn producing_block_after_genesis(&self) -> bool {
395        #[allow(clippy::arithmetic_side_effects)]
396        let two_mins_in_future = self.time() + Duration::from_secs(2 * 60);
397        self.block_producer.with(false, |bp| {
398            bp.current.won_slot_should_produce(two_mins_in_future)
399        }) && self.genesis_block().is_some_and(|b| {
400            let slot = &b.consensus_state().curr_global_slot_since_hard_fork;
401            let epoch = slot
402                .slot_number
403                .as_u32()
404                .checked_div(slot.slots_per_epoch.as_u32())
405                .expect("division by 0");
406            self.current_epoch() <= Some(epoch)
407        })
408    }
409
410    pub fn prevalidate_block(
411        &self,
412        block: &ArcBlockWithHash,
413        allow_block_too_late: bool,
414    ) -> Result<(), BlockPrevalidationError> {
415        let Some((genesis, cur_global_slot)) =
416            None.or_else(|| Some((self.genesis_block()?, self.cur_global_slot()?)))
417        else {
418            // we don't have genesis block. This should be impossible
419            // because we don't even know chain_id before we have genesis
420            // block, so we can't be connected to any peers from which
421            // we would receive a block.
422            bug_condition!("Tried to prevalidate a block before the genesis block was ready");
423            return Err(BlockPrevalidationError::GenesisNotReady);
424        };
425
426        prevalidate_block(block, &genesis, cur_global_slot, allow_block_too_late)
427    }
428
429    pub fn should_log_node_id(&self) -> bool {
430        self.config.testing_run
431    }
432
433    pub fn consensus_time_now(&self) -> Option<ConsensusTime> {
434        let (start_time, end_time) = self.slot_time(self.cur_global_slot()?.into())?;
435        let epoch = self.current_epoch()?;
436        let global_slot = self.cur_global_slot()?;
437        let slot = self.current_slot()?;
438        Some(ConsensusTime {
439            start_time,
440            end_time,
441            epoch,
442            global_slot,
443            slot,
444        })
445    }
446
447    pub fn consensus_time_best_tip(&self) -> Option<ConsensusTime> {
448        let best_tip = self.transition_frontier.best_tip()?;
449        let global_slot = best_tip
450            .curr_global_slot_since_hard_fork()
451            .slot_number
452            .as_u32();
453        let (start_time, end_time) = self.slot_time(global_slot.into())?;
454        let epoch = best_tip.consensus_state().epoch_count.as_u32();
455        let slot = best_tip.slot();
456        Some(ConsensusTime {
457            start_time,
458            end_time,
459            epoch,
460            global_slot,
461            slot,
462        })
463    }
464}
465
466#[serde_with::serde_as]
467#[derive(Debug, Clone, Serialize, Deserialize, MallocSizeOf)]
468#[expect(
469    clippy::large_enum_variant,
470    reason = "Doesn't make sense moving redux state onto heap"
471)]
472pub enum P2p {
473    Pending(#[ignore_malloc_size_of = "constant"] P2pConfig),
474    Ready(P2pState),
475}
476
477#[derive(Debug, thiserror::Error)]
478pub enum P2pInitializationError {
479    #[error("p2p is already initialized")]
480    AlreadyInitialized,
481}
482
483#[macro_export]
484macro_rules! p2p_ready {
485    ($p2p:expr, $time:expr) => {
486        p2p_ready!($p2p, "", $time)
487    };
488    ($p2p:expr, $reason:expr, $time:expr) => {
489        match $p2p.ready() {
490            Some(v) => v,
491            None => {
492                //panic!("p2p is not ready: {:?}\nline: {}", $reason, line!());
493                mina_core::error!($time; "p2p is not initialized: {}", $reason);
494                return;
495            }
496        }
497    };
498}
499
500impl P2p {
501    pub fn config(&self) -> &P2pConfig {
502        match self {
503            P2p::Pending(config) => config,
504            P2p::Ready(p2p_state) => &p2p_state.config,
505        }
506    }
507
508    // TODO: add chain id
509    pub fn initialize(&mut self, chain_id: &ChainId) -> Result<(), P2pInitializationError> {
510        let P2p::Pending(config) = self else {
511            return Err(P2pInitializationError::AlreadyInitialized);
512        };
513
514        let callbacks = Self::p2p_callbacks();
515        *self = P2p::Ready(P2pState::new(config.clone(), callbacks, chain_id));
516        Ok(())
517    }
518
519    fn p2p_callbacks() -> P2pCallbacks {
520        P2pCallbacks {
521            on_p2p_channels_transaction_received: Some(redux::callback!(
522                on_p2p_channels_transaction_received((peer_id: PeerId, info: Box<TransactionInfo>)) -> crate::Action {
523                    TransactionPoolCandidateAction::InfoReceived {
524                        peer_id,
525                        info: *info,
526                    }
527                }
528            )),
529            on_p2p_channels_transactions_libp2p_received: Some(redux::callback!(
530                on_p2p_channels_transactions_libp2p_received((peer_id: PeerId, transactions: Vec<TransactionWithHash>, message_id: P2pNetworkPubsubMessageCacheId)) -> crate::Action {
531                    TransactionPoolCandidateAction::Libp2pTransactionsReceived {
532                        message_id,
533                        transactions,
534                        peer_id
535                    }
536                }
537            )),
538            on_p2p_channels_snark_job_commitment_received: Some(redux::callback!(
539                on_p2p_channels_snark_job_commitment_received((peer_id: PeerId, commitment: Box<SnarkJobCommitment>)) -> crate::Action {
540                    SnarkPoolAction::CommitmentAdd { commitment: *commitment, sender: peer_id }
541                }
542            )),
543            on_p2p_channels_snark_received: Some(redux::callback!(
544                on_p2p_channels_snark_received((peer_id: PeerId, snark: Box<SnarkInfo>)) -> crate::Action {
545                    SnarkPoolCandidateAction::InfoReceived { peer_id, info: *snark }
546                }
547            )),
548            on_p2p_channels_snark_libp2p_received: Some(redux::callback!(
549                on_p2p_channels_snark_libp2p_received((peer_id: PeerId, snark: Box<Snark>)) -> crate::Action {
550                    SnarkPoolCandidateAction::WorkFetchSuccess { peer_id, work: *snark }
551                }
552            )),
553            on_p2p_channels_streaming_rpc_ready: Some(redux::callback!(
554                on_p2p_channels_streaming_rpc_ready(_var: ()) -> crate::Action {
555                    P2pCallbacksAction::P2pChannelsStreamingRpcReady
556                }
557            )),
558            on_p2p_channels_best_tip_request_received: Some(redux::callback!(
559                on_p2p_channels_best_tip_request_received(peer_id: PeerId) -> crate::Action {
560                    P2pCallbacksAction::RpcRespondBestTip { peer_id }
561                }
562            )),
563            on_p2p_disconnection_finish: Some(redux::callback!(
564                on_p2p_disconnection_finish(peer_id: PeerId) -> crate::Action {
565                    P2pCallbacksAction::P2pDisconnection { peer_id }
566                }
567            )),
568            on_p2p_connection_outgoing_error: Some(redux::callback!(
569                on_p2p_connection_outgoing_error((rpc_id: RpcId, error: P2pConnectionOutgoingError)) -> crate::Action {
570                    RpcAction::P2pConnectionOutgoingError { rpc_id, error }
571                }
572            )),
573            on_p2p_connection_outgoing_success: Some(redux::callback!(
574                on_p2p_connection_outgoing_success(rpc_id: RpcId) -> crate::Action {
575                    RpcAction::P2pConnectionOutgoingSuccess { rpc_id }
576                }
577            )),
578            on_p2p_connection_incoming_error: Some(redux::callback!(
579                on_p2p_connection_incoming_error((rpc_id: RpcId, error: String)) -> crate::Action {
580                    RpcAction::P2pConnectionIncomingError { rpc_id, error }
581                }
582            )),
583            on_p2p_connection_incoming_success: Some(redux::callback!(
584                on_p2p_connection_incoming_success(rpc_id: RpcId) -> crate::Action {
585                    RpcAction::P2pConnectionIncomingSuccess { rpc_id }
586                }
587            )),
588            on_p2p_connection_incoming_answer_ready: Some(redux::callback!(
589                on_p2p_connection_incoming_answer_ready((rpc_id: RpcId, peer_id: PeerId, answer: P2pConnectionResponse)) -> crate::Action {
590                    RpcAction::P2pConnectionIncomingAnswerReady { rpc_id, answer, peer_id }
591                }
592            )),
593            on_p2p_peer_best_tip_update: Some(redux::callback!(
594                on_p2p_peer_best_tip_update(best_tip: BlockWithHash<Arc<v2::MinaBlockBlockStableV2>>) -> crate::Action {
595                    TransitionFrontierCandidateAction::P2pBestTipUpdate { best_tip }
596                }
597            )),
598            on_p2p_channels_rpc_ready: Some(redux::callback!(
599                on_p2p_channels_rpc_ready(peer_id: PeerId) -> crate::Action {
600                    P2pCallbacksAction::P2pChannelsRpcReady { peer_id }
601                }
602            )),
603            on_p2p_channels_rpc_timeout: Some(redux::callback!(
604                on_p2p_channels_rpc_timeout((peer_id: PeerId, id: P2pRpcId)) -> crate::Action {
605                    P2pCallbacksAction::P2pChannelsRpcTimeout { peer_id, id }
606                }
607            )),
608            on_p2p_channels_rpc_response_received: Some(redux::callback!(
609                on_p2p_channels_rpc_response_received((peer_id: PeerId, id: P2pRpcId, response: Option<Box<P2pRpcResponse>>)) -> crate::Action {
610                    P2pCallbacksAction::P2pChannelsRpcResponseReceived { peer_id, id, response }
611                }
612            )),
613            on_p2p_channels_rpc_request_received: Some(redux::callback!(
614                on_p2p_channels_rpc_request_received((peer_id: PeerId, id: P2pRpcId, request: Box<P2pRpcRequest>)) -> crate::Action {
615                    P2pCallbacksAction::P2pChannelsRpcRequestReceived { peer_id, id, request }
616                }
617            )),
618            on_p2p_channels_streaming_rpc_response_received: Some(redux::callback!(
619                on_p2p_channels_streaming_rpc_response_received((peer_id: PeerId, id: P2pRpcId, response: Option<P2pStreamingRpcResponseFull>)) -> crate::Action {
620                    P2pCallbacksAction::P2pChannelsStreamingRpcResponseReceived { peer_id, id, response }
621                }
622            )),
623            on_p2p_channels_streaming_rpc_timeout: Some(redux::callback!(
624                on_p2p_channels_streaming_rpc_timeout((peer_id: PeerId, id: P2pRpcId)) -> crate::Action {
625                    P2pCallbacksAction::P2pChannelsStreamingRpcTimeout { peer_id, id }
626                }
627            )),
628            on_p2p_pubsub_message_received: Some(redux::callback!(
629                on_p2p_pubsub_message_received((message_id: P2pNetworkPubsubMessageCacheId)) -> crate::Action{
630                    P2pCallbacksAction::P2pPubsubValidateMessage { message_id }
631                }
632            )),
633        }
634    }
635
636    pub fn ready(&self) -> Option<&P2pState> {
637        if let P2p::Ready(state) = self {
638            Some(state)
639        } else {
640            None
641        }
642    }
643
644    pub fn ready_mut(&mut self) -> Option<&mut P2pState> {
645        if let P2p::Ready(state) = self {
646            Some(state)
647        } else {
648            None
649        }
650    }
651
652    pub fn unwrap(&self) -> &P2pState {
653        self.ready().expect("p2p is not initialized")
654    }
655
656    pub fn is_enabled<T>(&self, action: &T, time: Timestamp) -> bool
657    where
658        T: EnablingCondition<P2pState>,
659    {
660        match self {
661            P2p::Pending(_) => false,
662            P2p::Ready(p2p_state) => action.is_enabled(p2p_state, time),
663        }
664    }
665
666    pub fn my_id(&self) -> PeerId {
667        match self {
668            P2p::Pending(config) => &config.identity_pub_key,
669            P2p::Ready(state) => &state.config.identity_pub_key,
670        }
671        .peer_id()
672    }
673
674    pub fn get_peer(&self, peer_id: &PeerId) -> Option<&P2pPeerState> {
675        self.ready().and_then(|p2p| p2p.peers.get(peer_id))
676    }
677
678    pub fn get_ready_peer(&self, peer_id: &PeerId) -> Option<&P2pPeerStatusReady> {
679        self.ready().and_then(|p2p| p2p.get_ready_peer(peer_id))
680    }
681
682    pub fn ready_peers(&self) -> Vec<PeerId> {
683        self.ready_peers_iter()
684            .map(|(peer_id, _)| *peer_id)
685            .collect()
686    }
687
688    pub fn ready_peers_iter(&self) -> ReadyPeersIter<'_> {
689        ReadyPeersIter::new(self)
690    }
691}
692
693#[derive(Debug, Clone)]
694pub struct ReadyPeersIter<'a>(Option<std::collections::btree_map::Iter<'a, PeerId, P2pPeerState>>);
695
696impl<'a> ReadyPeersIter<'a> {
697    fn new(p2p: &'a P2p) -> Self {
698        ReadyPeersIter(p2p.ready().map(|p2p| p2p.peers.iter()))
699    }
700}
701
702impl<'a> Iterator for ReadyPeersIter<'a> {
703    type Item = (&'a PeerId, &'a P2pPeerStatusReady);
704
705    fn next(&mut self) -> Option<Self::Item> {
706        let iter = self.0.as_mut()?;
707        Some(loop {
708            let (peer_id, state) = iter.next()?;
709            if let Some(ready) = state.status.as_ready() {
710                break (peer_id, ready);
711            }
712        })
713    }
714}
mina_node/state.rs

mina_node/
state.rs
