Ported reducer and p2p effects in node

node/src/action.rs

@@ -8,6 +8,7 @@ pub use crate::consensus::ConsensusAction;
 pub use crate::event_source::EventSourceAction;
 pub use crate::external_snark_worker::ExternalSnarkWorkerAction;
 pub use crate::ledger::LedgerAction;
+use crate::p2p::callbacks::P2pCallbacksAction;
 pub use crate::p2p::P2pAction;
 pub use crate::rpc::RpcAction;
 pub use crate::snark::SnarkAction;
@@ -32,6 +33,8 @@ pub enum Action {
     EventSource(EventSourceAction),
 
     P2p(P2pAction),
+    P2pCallbacks(P2pCallbacksAction),
+
     Ledger(LedgerAction),
     Snark(SnarkAction),
     Consensus(ConsensusAction),
@@ -85,6 +88,7 @@ impl redux::EnablingCondition<crate::State> for Action {
             Action::WatchedAccounts(a) => a.is_enabled(state, time),
             Action::TransactionPool(a) => a.is_enabled(state, time),
             Action::TransactionPoolEffect(a) => a.is_enabled(state, time),
+            Action::P2pCallbacks(a) => a.is_enabled(state, time),
         }
     }
 }

node/src/action_kind.rs

@@ -23,6 +23,7 @@ use crate::external_snark_worker::ExternalSnarkWorkerAction;
 use crate::ledger::read::LedgerReadAction;
 use crate::ledger::write::LedgerWriteAction;
 use crate::ledger::LedgerAction;
+use crate::p2p::callbacks::P2pCallbacksAction;
 use crate::p2p::channels::best_tip::P2pChannelsBestTipAction;
 use crate::p2p::channels::best_tip_effectful::P2pChannelsBestTipEffectfulAction;
 use crate::p2p::channels::rpc::P2pChannelsRpcAction;
@@ -147,6 +148,7 @@ pub enum ActionKind {
     ConsensusBlockSnarkVerifySuccess,
     ConsensusDetectForkRange,
     ConsensusLongRangeForkResolve,
+    ConsensusP2pBestTipUpdate,
     ConsensusPrune,
     ConsensusShortRangeForkResolve,
     EventSourceNewEvent,
@@ -174,6 +176,15 @@ pub enum ActionKind {
     LedgerWriteInit,
     LedgerWritePending,
     LedgerWriteSuccess,
+    P2pCallbacksP2pChannelsRpcReady,
+    P2pCallbacksP2pChannelsRpcRequestReceived,
+    P2pCallbacksP2pChannelsRpcResponseReceived,
+    P2pCallbacksP2pChannelsRpcTimeout,
+    P2pCallbacksP2pChannelsStreamingRpcReady,
+    P2pCallbacksP2pChannelsStreamingRpcResponseReceived,
+    P2pCallbacksP2pChannelsStreamingRpcTimeout,
+    P2pCallbacksP2pDisconnection,
+    P2pCallbacksRpcRespondBestTip,
     P2pChannelsBestTipInit,
     P2pChannelsBestTipPending,
     P2pChannelsBestTipReady,
@@ -427,6 +438,7 @@ pub enum ActionKind {
     RpcLedgerAccountsGetPending,
     RpcLedgerAccountsGetSuccess,
     RpcMessageProgressGet,
+    RpcP2pConnectionIncomingAnswerReady,
     RpcP2pConnectionIncomingError,
     RpcP2pConnectionIncomingInit,
     RpcP2pConnectionIncomingPending,
@@ -599,7 +611,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 493;
+    pub const COUNT: u16 = 504;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -614,6 +626,7 @@ impl ActionKindGet for Action {
             Self::CheckTimeouts(a) => a.kind(),
             Self::EventSource(a) => a.kind(),
             Self::P2p(a) => a.kind(),
+            Self::P2pCallbacks(a) => a.kind(),
             Self::Ledger(a) => a.kind(),
             Self::Snark(a) => a.kind(),
             Self::Consensus(a) => a.kind(),
@@ -664,6 +677,32 @@ impl ActionKindGet for P2pAction {
     }
 }
 
+impl ActionKindGet for P2pCallbacksAction {
+    fn kind(&self) -> ActionKind {
+        match self {
+            Self::P2pChannelsRpcReady { .. } => ActionKind::P2pCallbacksP2pChannelsRpcReady,
+            Self::P2pChannelsRpcTimeout { .. } => ActionKind::P2pCallbacksP2pChannelsRpcTimeout,
+            Self::P2pChannelsRpcResponseReceived { .. } => {
+                ActionKind::P2pCallbacksP2pChannelsRpcResponseReceived
+            }
+            Self::P2pChannelsRpcRequestReceived { .. } => {
+                ActionKind::P2pCallbacksP2pChannelsRpcRequestReceived
+            }
+            Self::P2pChannelsStreamingRpcReady => {
+                ActionKind::P2pCallbacksP2pChannelsStreamingRpcReady
+            }
+            Self::P2pChannelsStreamingRpcTimeout { .. } => {
+                ActionKind::P2pCallbacksP2pChannelsStreamingRpcTimeout
+            }
+            Self::P2pChannelsStreamingRpcResponseReceived { .. } => {
+                ActionKind::P2pCallbacksP2pChannelsStreamingRpcResponseReceived
+            }
+            Self::P2pDisconnection { .. } => ActionKind::P2pCallbacksP2pDisconnection,
+            Self::RpcRespondBestTip { .. } => ActionKind::P2pCallbacksRpcRespondBestTip,
+        }
+    }
+}
+
 impl ActionKindGet for LedgerAction {
     fn kind(&self) -> ActionKind {
         match self {
@@ -698,6 +737,7 @@ impl ActionKindGet for ConsensusAction {
             Self::ShortRangeForkResolve { .. } => ActionKind::ConsensusShortRangeForkResolve,
             Self::LongRangeForkResolve { .. } => ActionKind::ConsensusLongRangeForkResolve,
             Self::BestTipUpdate { .. } => ActionKind::ConsensusBestTipUpdate,
+            Self::P2pBestTipUpdate { .. } => ActionKind::ConsensusP2pBestTipUpdate,
             Self::Prune => ActionKind::ConsensusPrune,
         }
     }
@@ -857,6 +897,9 @@ impl ActionKindGet for RpcAction {
             Self::P2pConnectionIncomingRespond { .. } => {
                 ActionKind::RpcP2pConnectionIncomingRespond
             }
+            Self::P2pConnectionIncomingAnswerReady { .. } => {
+                ActionKind::RpcP2pConnectionIncomingAnswerReady
+            }
             Self::P2pConnectionIncomingError { .. } => ActionKind::RpcP2pConnectionIncomingError,
             Self::P2pConnectionIncomingSuccess { .. } => {
                 ActionKind::RpcP2pConnectionIncomingSuccess

node/src/consensus/consensus_actions.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use mina_p2p_messages::v2::{MinaBlockBlockStableV2, StateHash};
-use openmina_core::block::ArcBlockWithHash;
+use openmina_core::block::{ArcBlockWithHash, BlockWithHash};
 use openmina_core::{action_event, ActionEvent};
 use serde::{Deserialize, Serialize};
 use snark::block_verify::SnarkBlockVerifyError;
@@ -53,6 +53,9 @@ pub enum ConsensusAction {
     BestTipUpdate {
         hash: StateHash,
     },
+    P2pBestTipUpdate {
+        best_tip: BlockWithHash<Arc<MinaBlockBlockStableV2>>,
+    },
     Prune,
 }
 
@@ -146,6 +149,7 @@ impl redux::EnablingCondition<crate::State> for ConsensusAction {
                     .consensus
                     .is_candidate_decided_to_use_as_tip(hash)
             },
+            ConsensusAction::P2pBestTipUpdate { .. } => true,
             ConsensusAction::Prune => {
                 state.consensus.best_tip().is_some()
             },

node/src/consensus/consensus_reducer.rs

@@ -5,7 +5,14 @@ use openmina_core::{
 use snark::block_verify::{SnarkBlockVerifyAction, SnarkBlockVerifyError};
 
 use crate::{
-    transition_frontier::sync::TransitionFrontierSyncAction, Action, State, WatchedAccountsAction,
+    transition_frontier::sync::{
+        ledger::{
+            snarked::TransitionFrontierSyncLedgerSnarkedAction,
+            staged::TransitionFrontierSyncLedgerStagedAction,
+        },
+        TransitionFrontierSyncAction,
+    },
+    Action, State, WatchedAccountsAction,
 };
 
 use super::{
@@ -235,6 +242,18 @@ impl ConsensusState {
 
                 transition_frontier_new_best_tip_handler(global_state, dispatcher);
             }
+            ConsensusAction::P2pBestTipUpdate { best_tip } => {
+                let dispatcher = state_context.into_dispatcher();
+                dispatcher.push(ConsensusAction::BlockReceived {
+                    hash: best_tip.hash.clone(),
+                    block: best_tip.block.clone(),
+                    chain_proof: None,
+                });
+
+                dispatcher.push(TransitionFrontierSyncLedgerSnarkedAction::PeersQuery);
+                dispatcher.push(TransitionFrontierSyncLedgerStagedAction::PartsPeerFetchInit);
+                dispatcher.push(TransitionFrontierSyncAction::BlocksPeersQuery);
+            }
             ConsensusAction::Prune => {
                 let Some(best_tip_hash) = state.best_tip.clone() else {
                     return;

node/src/effects.rs

@@ -1,5 +1,4 @@
 use openmina_core::log::system_time;
-use p2p::p2p_timeout_effects;
 
 use crate::block_producer::{block_producer_effects, BlockProducerAction};
 use crate::event_source::event_source_effects;
@@ -37,8 +36,6 @@ pub fn effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta) {
             store.dispatch(ExternalSnarkWorkerAction::Start);
 
             if store.state().p2p.ready().is_some() {
-                p2p_timeout_effects(store, &meta);
-
                 p2p_request_best_tip_if_needed(store);
                 p2p_request_transactions_if_needed(store);
                 p2p_request_snarks_if_needed(store);
@@ -95,6 +92,9 @@ pub fn effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta) {
         Action::WatchedAccounts(_) => {
             // Handled by reducer
         }
+        Action::P2pCallbacks(_) => {
+            // Handled by reducer
+        }
     }
 }
 

node/src/p2p/callbacks/mod.rs

@@ -0,0 +1,3 @@
+mod p2p_callbacks_actions;
+pub use p2p_callbacks_actions::P2pCallbacksAction;
+mod p2p_callbacks_reducer;

node/src/p2p/callbacks/p2p_callbacks_actions.rs

@@ -0,0 +1,68 @@
+use openmina_core::ActionEvent;
+use p2p::{
+    channels::{
+        rpc::{P2pRpcId, P2pRpcRequest, P2pRpcResponse},
+        streaming_rpc::P2pStreamingRpcResponseFull,
+    },
+    PeerId,
+};
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone, ActionEvent)]
+#[action_event(level = debug)]
+pub enum P2pCallbacksAction {
+    P2pChannelsRpcReady {
+        peer_id: PeerId,
+    },
+    P2pChannelsRpcTimeout {
+        peer_id: PeerId,
+        id: P2pRpcId,
+    },
+    P2pChannelsRpcResponseReceived {
+        peer_id: PeerId,
+        id: P2pRpcId,
+        response: Option<Box<P2pRpcResponse>>,
+    },
+    P2pChannelsRpcRequestReceived {
+        peer_id: PeerId,
+        id: P2pRpcId,
+        request: Box<P2pRpcRequest>,
+    },
+
+    P2pChannelsStreamingRpcReady,
+    P2pChannelsStreamingRpcTimeout {
+        peer_id: PeerId,
+        id: P2pRpcId,
+    },
+    P2pChannelsStreamingRpcResponseReceived {
+        peer_id: PeerId,
+        id: P2pRpcId,
+        response: Option<P2pStreamingRpcResponseFull>,
+    },
+
+    P2pDisconnection {
+        peer_id: PeerId,
+    },
+    RpcRespondBestTip {
+        peer_id: PeerId,
+    },
+}
+
+impl redux::EnablingCondition<crate::State> for P2pCallbacksAction {
+    fn is_enabled(&self, state: &crate::State, _time: redux::Timestamp) -> bool {
+        match self {
+            P2pCallbacksAction::P2pChannelsRpcReady { .. } => true,
+            P2pCallbacksAction::P2pChannelsRpcTimeout { .. } => true,
+            P2pCallbacksAction::P2pChannelsRpcResponseReceived { .. } => true,
+            P2pCallbacksAction::P2pChannelsRpcRequestReceived { .. } => true,
+            P2pCallbacksAction::P2pChannelsStreamingRpcReady => true,
+            P2pCallbacksAction::P2pChannelsStreamingRpcTimeout { .. } => true,
+            P2pCallbacksAction::P2pChannelsStreamingRpcResponseReceived { .. } => true,
+            P2pCallbacksAction::P2pDisconnection { .. } => true,
+            // TODO: what if we don't have best tip?
+            P2pCallbacksAction::RpcRespondBestTip { .. } => {
+                state.transition_frontier.best_tip().is_some()
+            }
+        }
+    }
+}