Review fixes

Author: 0xMimir

p2p/src/lib.rs

@@ -4,11 +4,11 @@ pub mod connection;
 pub mod disconnection;
 pub mod discovery;
 pub mod identity;
-pub mod peer;
 use bootstrap::P2pNetworkKadBootstrapState;
 use channels::{
-    rpc::P2pChannelsRpcAction, snark::P2pChannelsSnarkAction,
-    transaction::P2pChannelsTransactionAction,
+    best_tip::P2pChannelsBestTipAction, rpc::P2pChannelsRpcAction, snark::P2pChannelsSnarkAction,
+    snark_job_commitment::P2pChannelsSnarkJobCommitmentAction,
+    streaming_rpc::P2pChannelsStreamingRpcAction, transaction::P2pChannelsTransactionAction,
 };
 use connection::incoming::P2pConnectionIncomingAction;
 use disconnection::P2pDisconnectionAction;
@@ -27,6 +27,9 @@ pub mod identify;
 pub mod network;
 pub use self::network::*;
 
+pub mod peer;
+pub use peer::*;
+
 mod p2p_config;
 pub use p2p_config::*;
 
@@ -117,5 +120,8 @@ pub trait P2pActionTrait<State>:
     + From<P2pChannelsRpcAction>
     + From<P2pDisconnectionAction>
     + From<P2pNetworkSchedulerEffectfulAction>
+    + From<P2pChannelsBestTipAction>
+    + From<P2pChannelsSnarkJobCommitmentAction>
+    + From<P2pChannelsStreamingRpcAction>
 {
 }

p2p/src/network/pnet/p2p_network_pnet_actions.rs

@@ -42,7 +42,11 @@ impl From<P2pNetworkPnetAction> for crate::P2pAction {
 }
 
 impl redux::EnablingCondition<P2pState> for P2pNetworkPnetAction {
-    fn is_enabled(&self, _state: &P2pState, _time: redux::Timestamp) -> bool {
-        true
+    fn is_enabled(&self, state: &P2pState, _time: redux::Timestamp) -> bool {
+        state
+            .network
+            .scheduler
+            .connection_state(self.addr())
+            .is_some()
     }
 }

p2p/src/network/pnet/p2p_network_pnet_reducer.rs

@@ -1,4 +1,4 @@
-use openmina_core::Substate;
+use openmina_core::{bug_condition, Substate};
 use salsa_simple::XSalsa20;
 
 use crate::{
@@ -68,6 +68,7 @@ impl P2pNetworkPnetState {
                     .pnet;
 
                 let Half::Done { to_send, .. } = &pnet_state.incoming else {
+                    // this should be only if incoming state didn't receive enough data
                     return Ok(());
                 };
 
@@ -93,6 +94,7 @@ impl P2pNetworkPnetState {
                     .pnet;
 
                 let Half::Done { to_send, .. } = &pnet_state.outgoing else {
+                    // this should be only if we haven't set enough data to change from `Buffering` to `Done`
                     return Ok(());
                 };
 
@@ -111,7 +113,21 @@ impl P2pNetworkPnetState {
             } => {
                 pnet_state.outgoing.reduce(&pnet_state.shared_secret, nonce);
 
-                let dispatcher = state_context.into_dispatcher();
+                let (dispatcher, state) = state_context.into_dispatcher_and_state();
+                let scheduler_state: &P2pNetworkSchedulerState = state.substate()?;
+                let pnet_state = &scheduler_state
+                    .connection_state(addr)
+                    .ok_or_else(|| format!("Missing connection for action: {:?}", action))?
+                    .pnet;
+
+                if !matches!(&pnet_state.outgoing, Half::Done { .. }) {
+                    bug_condition!(
+                        "Invalid state for `P2pNetworkPnetAction::SetupNonce`, state: {:?}",
+                        pnet_state.outgoing
+                    );
+                    return Ok(());
+                };
+
                 dispatcher.push(P2pNetworkPnetEffectfulAction::SetupNonce {
                     addr: *addr,
                     nonce: nonce.clone(),

p2p/src/network/scheduler/p2p_network_scheduler_actions.rs

@@ -127,8 +127,7 @@ impl redux::EnablingCondition<P2pState> for P2pNetworkSchedulerAction {
             | P2pNetworkSchedulerAction::ListenerReady { .. }
             | P2pNetworkSchedulerAction::ListenerError { .. }
             | P2pNetworkSchedulerAction::SelectDone { .. }
-            | P2pNetworkSchedulerAction::SelectError { .. }
-            | P2pNetworkSchedulerAction::YamuxDidInit { .. } => true,
+            | P2pNetworkSchedulerAction::SelectError { .. } => true,
             P2pNetworkSchedulerAction::IncomingDidAccept { addr, .. } => {
                 addr.as_ref().map_or(false, |addr| {
                     !state.network.scheduler.connections.contains_key(addr)
@@ -149,7 +148,8 @@ impl redux::EnablingCondition<P2pState> for P2pNetworkSchedulerAction {
                 .get(addr)
                 .map_or(false, |conn_state| !conn_state.incoming),
             P2pNetworkSchedulerAction::IncomingDataDidReceive { addr, .. }
-            | P2pNetworkSchedulerAction::IncomingDataIsReady { addr } => {
+            | P2pNetworkSchedulerAction::IncomingDataIsReady { addr }
+            | P2pNetworkSchedulerAction::YamuxDidInit { addr, .. } => {
                 state.network.scheduler.connections.contains_key(addr)
             }
             P2pNetworkSchedulerAction::Disconnect { addr, .. }