Ported effects

Author: 0xMimir

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,13 @@ pub enum ActionKind {
     LedgerWriteInit,
     LedgerWritePending,
     LedgerWriteSuccess,
+    P2pCallbacksP2pChannelsRpcReady,
+    P2pCallbacksP2pChannelsRpcRequestReceived,
+    P2pCallbacksP2pChannelsRpcResponseReceived,
+    P2pCallbacksP2pChannelsRpcTimeout,
+    P2pCallbacksP2pChannelsStreamingRpcReady,
+    P2pCallbacksP2pChannelsStreamingRpcResponseReceived,
+    P2pCallbacksP2pChannelsStreamingRpcTimeout,
     P2pChannelsBestTipInit,
     P2pChannelsBestTipPending,
     P2pChannelsBestTipReady,
@@ -427,6 +436,7 @@ pub enum ActionKind {
     RpcLedgerAccountsGetPending,
     RpcLedgerAccountsGetSuccess,
     RpcMessageProgressGet,
+    RpcP2pConnectionIncomingAnswerReady,
     RpcP2pConnectionIncomingError,
     RpcP2pConnectionIncomingInit,
     RpcP2pConnectionIncomingPending,
@@ -510,6 +520,7 @@ pub enum ActionKind {
     TransactionPoolVerifyError,
     TransactionPoolEffectfulFetchAccounts,
     TransitionFrontierGenesisInject,
+    TransitionFrontierRpcRespondBestTip,
     TransitionFrontierSyncFailed,
     TransitionFrontierSynced,
     TransitionFrontierGenesisLedgerLoadInit,
@@ -559,6 +570,7 @@ pub enum ActionKind {
     TransitionFrontierSyncLedgerSnarkedNumAccountsReceived,
     TransitionFrontierSyncLedgerSnarkedNumAccountsRejected,
     TransitionFrontierSyncLedgerSnarkedNumAccountsSuccess,
+    TransitionFrontierSyncLedgerSnarkedP2pDisconnection,
     TransitionFrontierSyncLedgerSnarkedPeerQueryAddressError,
     TransitionFrontierSyncLedgerSnarkedPeerQueryAddressInit,
     TransitionFrontierSyncLedgerSnarkedPeerQueryAddressPending,
@@ -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,30 @@ 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
+            }
+        }
+    }
+}
+
 impl ActionKindGet for LedgerAction {
     fn kind(&self) -> ActionKind {
         match self {
@@ -698,6 +735,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,
         }
     }
@@ -712,6 +750,7 @@ impl ActionKindGet for TransitionFrontierAction {
             Self::GenesisInject => ActionKind::TransitionFrontierGenesisInject,
             Self::Synced { .. } => ActionKind::TransitionFrontierSynced,
             Self::SyncFailed { .. } => ActionKind::TransitionFrontierSyncFailed,
+            Self::RpcRespondBestTip { .. } => ActionKind::TransitionFrontierRpcRespondBestTip,
         }
     }
 }
@@ -857,6 +896,9 @@ impl ActionKindGet for RpcAction {
             Self::P2pConnectionIncomingRespond { .. } => {
                 ActionKind::RpcP2pConnectionIncomingRespond
             }
+            Self::P2pConnectionIncomingAnswerReady { .. } => {
+                ActionKind::RpcP2pConnectionIncomingAnswerReady
+            }
             Self::P2pConnectionIncomingError { .. } => ActionKind::RpcP2pConnectionIncomingError,
             Self::P2pConnectionIncomingSuccess { .. } => {
                 ActionKind::RpcP2pConnectionIncomingSuccess
@@ -1880,6 +1922,9 @@ impl ActionKindGet for TransitionFrontierSyncLedgerSnarkedAction {
                 ActionKind::TransitionFrontierSyncLedgerSnarkedMerkleTreeSyncSuccess
             }
             Self::Success => ActionKind::TransitionFrontierSyncLedgerSnarkedSuccess,
+            Self::P2pDisconnection { .. } => {
+                ActionKind::TransitionFrontierSyncLedgerSnarkedP2pDisconnection
+            }
         }
     }
 }

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,7 @@ 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 +235,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

@@ -92,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,48 @@
+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>,
+    },
+}
+
+impl redux::EnablingCondition<crate::State> for P2pCallbacksAction {
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        true
+    }
+}