feat(actions): time in enabling conditions

Author: akoptelov

node/src/p2p/channels/rpc/p2p_channels_rpc_actions.rs

@@ -2,12 +2,6 @@ use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pChannelsRpcAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
-        match self {
-            Self::Timeout { peer_id, id } => {
-                self.is_enabled(&state.p2p)
-                    && state.p2p.is_peer_rpc_timed_out(peer_id, *id, state.time())
-            }
-            _ => self.is_enabled(&state.p2p),
-        }
+        self.is_enabled(&state.p2p)
     }
 }

node/src/p2p/connection/incoming/p2p_connection_incoming_actions.rs

@@ -2,15 +2,6 @@ use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pConnectionIncomingAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
-        match self {
-            Self::Timeout { peer_id } => {
-                let peer = state.p2p.peers.get(peer_id);
-                let timed_out = peer
-                    .and_then(|peer| peer.status.as_connecting()?.as_incoming())
-                    .map_or(false, |s| s.is_timed_out(state.time()));
-                timed_out && self.is_enabled(&state.p2p)
-            }
-            _ => self.is_enabled(&state.p2p),
-        }
+        self.is_enabled(&state.p2p)
     }
 }

node/src/p2p/connection/outgoing/p2p_connection_outgoing_actions.rs

@@ -1,40 +1,7 @@
-use std::time::Duration;
-
-use crate::p2p::connection::P2pConnectionState;
-use crate::p2p::P2pPeerStatus;
-
 use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pConnectionOutgoingAction {
     fn is_enabled(&self, state: &crate::State) -> bool {
-        match self {
-            P2pConnectionOutgoingAction::Reconnect { opts, .. } => {
-                if !self.is_enabled(&state.p2p) {
-                    return false;
-                }
-
-                let Some(peer) = state.p2p.peers.get(opts.peer_id()) else {
-                    return false;
-                };
-                let delay_passed = match &peer.status {
-                    P2pPeerStatus::Connecting(P2pConnectionState::Outgoing(
-                        P2pConnectionOutgoingState::Error { time, .. },
-                    ))
-                    | P2pPeerStatus::Disconnected { time, .. } => {
-                        state.time().checked_sub(*time) >= Some(Duration::from_secs(30))
-                    }
-                    _ => true,
-                };
-                delay_passed
-            }
-            P2pConnectionOutgoingAction::Timeout { peer_id } => {
-                let peer = state.p2p.peers.get(peer_id);
-                let timed_out = peer
-                    .and_then(|peer| peer.status.as_connecting()?.as_outgoing())
-                    .map_or(false, |s| s.is_timed_out(state.time()));
-                timed_out && self.is_enabled(&state.p2p)
-            }
-            _ => self.is_enabled(&state.p2p),
-        }
+        self.is_enabled(&state.p2p)
     }
 }

p2p/src/channels/rpc/p2p_channels_rpc_actions.rs

@@ -1,3 +1,4 @@
+use redux::Timestamp;
 use serde::{Deserialize, Serialize};
 
 use crate::{P2pState, PeerId};
@@ -61,7 +62,7 @@ impl P2pChannelsRpcAction {
 }
 
 impl redux::EnablingCondition<P2pState> for P2pChannelsRpcAction {
-    fn is_enabled(&self, state: &P2pState) -> bool {
+    fn is_enabled_with_time(&self, state: &P2pState, time: Timestamp) -> bool {
         match self {
             P2pChannelsRpcAction::Init { peer_id } => {
                 state.get_ready_peer(peer_id).map_or(false, |p| {
@@ -93,12 +94,8 @@ impl redux::EnablingCondition<P2pState> for P2pChannelsRpcAction {
                     })
             },
             P2pChannelsRpcAction::Timeout { peer_id, id } => {
-                state.get_ready_peer(peer_id).map_or(false, |p| match &p.channels.rpc {
-                    P2pChannelsRpcState::Ready { local, .. } => {
-                        matches!(local, P2pRpcLocalState::Requested { id: rpc_id, .. } if rpc_id == id)
-                    },
-                    _ => false,
-                })
+                state.get_ready_peer(peer_id).map_or(false, |p| matches!(&p.channels.rpc, P2pChannelsRpcState::Ready { local: P2pRpcLocalState::Requested { id: rpc_id, .. }, .. } if rpc_id == id))
+                    && state.is_peer_rpc_timed_out(peer_id, *id, time)
             },
             P2pChannelsRpcAction::ResponseReceived { peer_id, id, .. } => {
                 // TODO(binier): use consensus to enforce that peer doesn't send

p2p/src/connection/incoming/p2p_connection_incoming_actions.rs

@@ -79,7 +79,7 @@ impl P2pConnectionIncomingAction {
 }
 
 impl redux::EnablingCondition<P2pState> for P2pConnectionIncomingAction {
-    fn is_enabled(&self, state: &P2pState) -> bool {
+    fn is_enabled_with_time(&self, state: &P2pState, time: redux::Timestamp) -> bool {
         match self {
             P2pConnectionIncomingAction::Init { opts, .. } => {
                 state.incoming_accept(opts.peer_id, &opts.offer).is_ok()
@@ -160,7 +160,7 @@ impl redux::EnablingCondition<P2pState> for P2pConnectionIncomingAction {
                 .peers
                 .get(peer_id)
                 .and_then(|peer| peer.status.as_connecting()?.as_incoming())
-                .is_some(),
+                .map_or(false, |s| s.is_timed_out(time)),
             P2pConnectionIncomingAction::Error { peer_id, error } => state
                 .peers
                 .get(peer_id)

p2p/src/connection/outgoing/p2p_connection_outgoing_actions.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use serde::{Deserialize, Serialize};
 
 use openmina_core::requests::RpcId;
@@ -97,7 +99,7 @@ impl P2pConnectionOutgoingAction {
 }
 
 impl redux::EnablingCondition<P2pState> for P2pConnectionOutgoingAction {
-    fn is_enabled(&self, state: &P2pState) -> bool {
+    fn is_enabled_with_time(&self, state: &P2pState, time: redux::Timestamp) -> bool {
         match self {
             P2pConnectionOutgoingAction::RandomInit => {
                 !state.already_has_min_peers() && !state.initial_unused_peers().is_empty()
@@ -130,7 +132,9 @@ impl redux::EnablingCondition<P2pState> for P2pConnectionOutgoingAction {
                     .min_by_key(|(time, ..)| *time)
                     .filter(|(_, id, _)| *id == opts.peer_id())
                     .filter(|(.., peer_opts)| peer_opts.as_ref().map_or(true, |o| o == opts))
-                    .is_some()
+                    .map_or(false, |(t, ..)| {
+                        time.checked_sub(t) >= Some(Duration::from_secs(30))
+                    })
             }
             P2pConnectionOutgoingAction::OfferSdpCreatePending { peer_id } => state
                 .peers
@@ -237,7 +241,7 @@ impl redux::EnablingCondition<P2pState> for P2pConnectionOutgoingAction {
                 .peers
                 .get(peer_id)
                 .and_then(|peer| peer.status.as_connecting()?.as_outgoing())
-                .is_some(),
+                .map_or(false, |s| s.is_timed_out(time)),
             P2pConnectionOutgoingAction::Error { peer_id, error } => state
                 .peers
                 .get(peer_id)