Merge pull request #944 from 0xMimir/feat/add-remaining-callbacks

Ported remaining callbacks in p2p layer and snark layer

Author: 0xMimir

node/src/action_kind.rs

@@ -351,7 +351,8 @@ pub enum ActionKind {
     P2pNetworkIdentifyStreamNew,
     P2pNetworkIdentifyStreamPrune,
     P2pNetworkIdentifyStreamRemoteClose,
-    P2pNetworkIdentifyStreamEffectfulSendIdentify,
+    P2pNetworkIdentifyStreamSendIdentify,
+    P2pNetworkIdentifyStreamEffectfulGetListenAddresses,
     P2pNetworkKadBootstrapAppendRequest,
     P2pNetworkKadBootstrapCreateRequests,
     P2pNetworkKadBootstrapFinalizeRequests,
@@ -435,20 +436,17 @@ pub enum ActionKind {
     P2pNetworkSchedulerOutgoingDidConnect,
     P2pNetworkSchedulerPrune,
     P2pNetworkSchedulerPruneStream,
-    P2pNetworkSchedulerPruneStreams,
     P2pNetworkSchedulerSelectDone,
     P2pNetworkSchedulerSelectError,
     P2pNetworkSchedulerYamuxDidInit,
     P2pNetworkSchedulerEffectfulDisconnect,
-    P2pNetworkSchedulerEffectfulError,
     P2pNetworkSchedulerEffectfulIncomingConnectionIsReady,
     P2pNetworkSchedulerEffectfulIncomingDataIsReady,
     P2pNetworkSchedulerEffectfulIncomingDidAccept,
     P2pNetworkSchedulerEffectfulInterfaceDetected,
     P2pNetworkSchedulerEffectfulNoiseSelectDone,
     P2pNetworkSchedulerEffectfulOutgoingConnect,
     P2pNetworkSchedulerEffectfulOutgoingDidConnect,
-    P2pNetworkSchedulerEffectfulSelectError,
     P2pNetworkSelectIncomingData,
     P2pNetworkSelectIncomingDataAuth,
     P2pNetworkSelectIncomingDataMux,
@@ -688,7 +686,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 579;
+    pub const COUNT: u16 = 577;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1804,7 +1802,6 @@ impl ActionKindGet for P2pNetworkSchedulerAction {
             Self::Error { .. } => ActionKind::P2pNetworkSchedulerError,
             Self::Disconnected { .. } => ActionKind::P2pNetworkSchedulerDisconnected,
             Self::Prune { .. } => ActionKind::P2pNetworkSchedulerPrune,
-            Self::PruneStreams { .. } => ActionKind::P2pNetworkSchedulerPruneStreams,
             Self::PruneStream { .. } => ActionKind::P2pNetworkSchedulerPruneStream,
         }
     }
@@ -1925,7 +1922,7 @@ impl ActionKindGet for P2pNetworkRpcAction {
 impl ActionKindGet for P2pConnectionOutgoingEffectfulAction {
     fn kind(&self) -> ActionKind {
         match self {
-            Self::RandomInit => ActionKind::P2pConnectionOutgoingEffectfulRandomInit,
+            Self::RandomInit { .. } => ActionKind::P2pConnectionOutgoingEffectfulRandomInit,
             Self::Init { .. } => ActionKind::P2pConnectionOutgoingEffectfulInit,
             Self::OfferSend { .. } => ActionKind::P2pConnectionOutgoingEffectfulOfferSend,
             Self::AnswerSet { .. } => ActionKind::P2pConnectionOutgoingEffectfulAnswerSet,
@@ -1973,9 +1970,7 @@ impl ActionKindGet for P2pNetworkSchedulerEffectfulAction {
                 ActionKind::P2pNetworkSchedulerEffectfulIncomingDataIsReady
             }
             Self::NoiseSelectDone { .. } => ActionKind::P2pNetworkSchedulerEffectfulNoiseSelectDone,
-            Self::SelectError { .. } => ActionKind::P2pNetworkSchedulerEffectfulSelectError,
             Self::Disconnect { .. } => ActionKind::P2pNetworkSchedulerEffectfulDisconnect,
-            Self::Error { .. } => ActionKind::P2pNetworkSchedulerEffectfulError,
         }
     }
 }
@@ -2034,6 +2029,7 @@ impl ActionKindGet for P2pNetworkIdentifyStreamAction {
             Self::Close { .. } => ActionKind::P2pNetworkIdentifyStreamClose,
             Self::RemoteClose { .. } => ActionKind::P2pNetworkIdentifyStreamRemoteClose,
             Self::Prune { .. } => ActionKind::P2pNetworkIdentifyStreamPrune,
+            Self::SendIdentify { .. } => ActionKind::P2pNetworkIdentifyStreamSendIdentify,
         }
     }
 }
@@ -2102,7 +2098,9 @@ impl ActionKindGet for P2pNetworkKademliaStreamAction {
 impl ActionKindGet for P2pNetworkIdentifyStreamEffectfulAction {
     fn kind(&self) -> ActionKind {
         match self {
-            Self::SendIdentify { .. } => ActionKind::P2pNetworkIdentifyStreamEffectfulSendIdentify,
+            Self::GetListenAddresses { .. } => {
+                ActionKind::P2pNetworkIdentifyStreamEffectfulGetListenAddresses
+            }
         }
     }
 }

node/src/rpc/rpc_reducer.rs

@@ -1,11 +1,12 @@
 use openmina_core::{
     block::AppliedBlock,
     bug_condition,
-    requests::{RequestId, RpcIdType},
+    requests::{RequestId, RpcId, RpcIdType},
 };
 use p2p::{
     connection::{incoming::P2pConnectionIncomingAction, outgoing::P2pConnectionOutgoingAction},
     webrtc::P2pConnectionResponse,
+    PeerId,
 };
 use redux::ActionWithMeta;
 
@@ -79,11 +80,20 @@ impl RpcState {
                 state.requests.insert(*rpc_id, rpc_state);
 
                 let dispatcher = state_context.into_dispatcher();
+
                 dispatcher.push(P2pConnectionOutgoingAction::Init {
                     opts: opts.clone(),
                     rpc_id: Some(*rpc_id),
+                    on_success: Some(redux::callback!(
+                        on_p2p_connection_outgoing_rpc_connection_success((peer_id: PeerId, rpc_id: Option<RpcId>)) -> crate::Action {
+                            let Some(rpc_id) = rpc_id else {
+                                unreachable!("RPC ID not provided");
+                            };
+
+                            RpcAction::P2pConnectionOutgoingPending{ rpc_id }
+                        }
+                    )),
                 });
-                dispatcher.push(RpcAction::P2pConnectionOutgoingPending { rpc_id: *rpc_id });
             }
             RpcAction::P2pConnectionOutgoingPending { rpc_id } => {
                 let Some(rpc) = state.requests.get_mut(rpc_id) else {

node/src/snark/snark_effects.rs

@@ -1,6 +1,4 @@
-use snark::work_verify::SnarkWorkVerifyAction;
-
-use crate::{snark_pool::candidate::SnarkPoolCandidateAction, Service, Store};
+use crate::{Service, Store};
 
 use super::{SnarkAction, SnarkActionWithMeta};
 
@@ -12,35 +10,7 @@ pub fn snark_effects<S: Service>(store: &mut Store<S>, action: SnarkActionWithMe
         SnarkAction::BlockVerifyEffect(a) => {
             a.effects(&meta, store);
         }
-        SnarkAction::WorkVerify(a) => match a {
-            // TODO(tizoc): handle this logic with the on_error callback passed on the Init action
-            SnarkWorkVerifyAction::Error { req_id, .. } => {
-                let req = store.state().snark.work_verify.jobs.get(req_id);
-                let Some(req) = req else { return };
-                let sender = req.sender().parse().unwrap();
-
-                store.dispatch(SnarkPoolCandidateAction::WorkVerifyError {
-                    peer_id: sender,
-                    verify_id: req_id,
-                });
-            }
-            // TODO(tizoc): handle this logic with the on_success callback passed on the Init action
-            SnarkWorkVerifyAction::Success { req_id } => {
-                let req = store.state().snark.work_verify.jobs.get(req_id);
-                let Some(req) = req else { return };
-                let sender = req.sender().parse().unwrap();
-                let batch = req.batch().to_vec();
-
-                store.dispatch(SnarkPoolCandidateAction::WorkVerifySuccess {
-                    peer_id: sender,
-                    verify_id: req_id,
-                    batch,
-                });
-            }
-            SnarkWorkVerifyAction::Init { .. } => {}
-            SnarkWorkVerifyAction::Pending { .. } => {}
-            SnarkWorkVerifyAction::Finish { .. } => {}
-        },
+        SnarkAction::WorkVerify(_) => {}
         SnarkAction::WorkVerifyEffect(a) => {
             a.effects(&meta, store);
         }

p2p/src/channels/signaling/discovery/p2p_channels_signaling_discovery_reducer.rs

@@ -269,6 +269,7 @@ impl P2pChannelsSignalingDiscoveryState {
                         },
                     },
                     rpc_id: None,
+                    on_success: None,
                 };
                 let accepted = redux::EnablingCondition::is_enabled(&action, state, meta.time());
                 if accepted {

p2p/src/connection/outgoing/p2p_connection_outgoing_actions.rs

@@ -1,4 +1,5 @@
 use openmina_core::ActionEvent;
+use redux::Callback;
 use serde::{Deserialize, Serialize};
 
 use openmina_core::requests::RpcId;
@@ -18,6 +19,7 @@ pub enum P2pConnectionOutgoingAction {
     Init {
         opts: P2pConnectionOutgoingInitOpts,
         rpc_id: Option<RpcId>,
+        on_success: Option<Callback<(PeerId, Option<RpcId>)>>,
     },
     /// Reconnect to an existing peer.
     // TODO: rename `Init` and `Reconnect` to `New` and `Connect` or something