Move time-related enabling conditions to p2p

ARCHITECTURE.md

@@ -33,14 +33,14 @@ pub trait EnablingCondition<State> {
     /// Enabling condition for the Action.
     ///
     /// Checks if the given action is enabled for a given state.
-    fn is_enabled(&self, state: &State) -> bool {
+    fn is_enabled(&self, state: &State, time: Timestamp) -> bool {
         ...
     }
 }
 ```
 
-`is_enabled(state)` must return `false`, if action doesn't make sense given
-the current state.
+`is_enabled(state, time)` must return `false`, if action doesn't make sense given
+the current state and, optionally, time.
 
 For example message action from peer that isn't connected or we don't know
 about in the state, must not be enabled.

Cargo.toml

@@ -38,6 +38,7 @@ mina-poseidon = {git = "https://github.com/openmina/proof-systems", branch = "le
 poly-commitment = {git = "https://github.com/openmina/proof-systems", branch = "ledger-newtypes-rampup4-vrf"}
 libp2p = { git = "https://github.com/openmina/rust-libp2p", rev = "cd5425a759d959d7fde58a42f71ab059449760c5", default-features = false }
 vrf = { path = "vrf" }
+redux = { git = "https://github.com/openmina/redux-rs.git", branch = "feat/enabling-condition-with-time", features = ["serde"] }
 
 [profile.fuzz]
 inherits = "release"

cli/Cargo.toml

@@ -17,7 +17,7 @@ num_cpus = "1.0"
 rayon = "1.5"
 tokio = { version = "1.26.0" }
 libp2p-identity = { version = "=0.2.7", features = ["peerid"] }
-redux = { git = "https://github.com/openmina/redux-rs.git", branch="feat/global-time", features = ["serde"] }
+redux = { workspace = true }
 ledger = { workspace = true }
 mina-p2p-messages = { workspace = true }
 vrf = { workspace = true }

core/Cargo.toml

@@ -9,7 +9,7 @@ serde = { version = "1.0.147", features = ["rc"] }
 slab = { version = "0.4.7", features = ["serde"] }
 tracing = { version = "0.1", features = ["std"] }
 sha2 = "0.10.6"
-redux = { git = "https://github.com/openmina/redux-rs.git", branch="feat/global-time", features = ["serde"] }
+redux = { workspace = true }
 tokio = { version = "1.26", features = ["sync"] }
 
 mina-hasher = { workspace = true }

node/Cargo.toml

@@ -18,7 +18,7 @@ bs58 = "0.4.0"
 bincode = "1.3.3"
 hex = "0.4.3"
 rand = "0.8"
-redux = { git = "https://github.com/openmina/redux-rs.git", branch="feat/global-time", features = ["serde"] }
+redux = { workspace = true }
 mina-hasher = { workspace = true }
 mina-signer = { workspace = true }
 ledger = { workspace = true }

node/invariants/Cargo.toml

@@ -15,5 +15,5 @@ strum = "0.24.1"
 strum_macros = "0.24.3"
 serde = "1.0.147"
 serde_json = { version = "1.0.82", features = ["unbounded_depth", "arbitrary_precision"] }
-redux = { git = "https://github.com/openmina/redux-rs.git", branch="feat/global-time", features = ["serde"] }
+redux = { workspace = true }
 node = { path = "../" }

node/native/Cargo.toml

@@ -14,7 +14,7 @@ warp = "0.3"
 libp2p = { workspace = true, features = ["macros", "serde", "tcp", "dns", "tokio", "yamux", "pnet", "noise", "gossipsub"] }
 juniper = { version = "0.15.11" }
 juniper_warp = { version = "0.7.0" }
-redux = { git = "https://github.com/openmina/redux-rs.git", branch="feat/global-time", features = ["serde"] }
+redux = { workspace = true }
 ledger = { workspace = true }
 mina-p2p-messages = { workspace = true }
 mina-signer = { workspace = true }

node/src/action.rs

@@ -51,7 +51,7 @@ impl redux::EnablingCondition<crate::State> for CheckTimeoutsAction {}
 
 #[cfg(feature = "replay")]
 impl redux::EnablingCondition<crate::State> for Action {
-    fn is_enabled(&self, _: &crate::State) -> bool {
+    fn is_enabled(&self, _: &crate::State, _time: redux::Timestamp) -> bool {
         true
     }
 }

node/src/block_producer/block_producer_actions.rs

@@ -41,9 +41,9 @@ pub enum BlockProducerAction {
 }
 
 impl redux::EnablingCondition<crate::State> for BlockProducerAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
         match self {
-            BlockProducerAction::VrfEvaluator(a) => a.is_enabled(state),
+            BlockProducerAction::VrfEvaluator(a) => a.is_enabled(state, time),
             BlockProducerAction::BestTipUpdate { .. } => true,
             BlockProducerAction::WonSlotSearch => state
                 .block_producer
@@ -66,12 +66,12 @@ impl redux::EnablingCondition<crate::State> for BlockProducerAction {
                     && won_slot.global_slot() >= state.cur_global_slot().unwrap()
                     && won_slot > best_tip
             }),
-            BlockProducerAction::WonSlotWait => state.block_producer.with(false, |this| {
-                this.current.won_slot_should_wait(state.time())
-            }),
-            BlockProducerAction::WonSlotProduceInit => state.block_producer.with(false, |this| {
-                this.current.won_slot_should_produce(state.time())
-            }),
+            BlockProducerAction::WonSlotWait => state
+                .block_producer
+                .with(false, |this| this.current.won_slot_should_wait(time)),
+            BlockProducerAction::WonSlotProduceInit => state
+                .block_producer
+                .with(false, |this| this.current.won_slot_should_produce(time)),
             BlockProducerAction::StagedLedgerDiffCreateInit => {
                 state.block_producer.with(false, |this| {
                     matches!(

node/src/block_producer/vrf_evaluator/block_producer_vrf_evaluator_actions.rs

@@ -43,7 +43,7 @@ pub enum BlockProducerVrfEvaluatorAction {
 }
 
 impl redux::EnablingCondition<crate::State> for BlockProducerVrfEvaluatorAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
+    fn is_enabled(&self, state: &crate::State, _time: redux::Timestamp) -> bool {
         match self {
             BlockProducerVrfEvaluatorAction::UpdateProducerAndDelegates { .. } => {
                 state.block_producer.with(false, |this| {

node/src/consensus/consensus_actions.rs

@@ -44,7 +44,7 @@ pub enum ConsensusAction {
 }
 
 impl redux::EnablingCondition<crate::State> for ConsensusAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
+    fn is_enabled(&self, state: &crate::State, _time: redux::Timestamp) -> bool {
         match self {
             ConsensusAction::BlockReceived { hash, .. } => {
                 !state.consensus.blocks.contains_key(hash)

node/src/effects.rs

@@ -97,7 +97,7 @@ pub fn effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta) {
             #[cfg(feature = "p2p-libp2p")]
             {
                 let state = store.state();
-                for (peer_id, id) in state.p2p.peer_rpc_timeouts(state.time()) {
+                for (peer_id, id) in state.p2p.peer_rpc_timeouts(meta.prev_time()) {
                     store.dispatch(P2pChannelsRpcAction::Timeout { peer_id, id });
                 }
             }

node/src/event_source/event_source_actions.rs

@@ -24,7 +24,7 @@ pub enum EventSourceAction {
 }
 
 impl redux::EnablingCondition<crate::State> for EventSourceAction {
-    fn is_enabled(&self, _: &crate::State) -> bool {
+    fn is_enabled(&self, _: &crate::State, _time: redux::Timestamp) -> bool {
         match self {
             EventSourceAction::ProcessEvents => true,
             EventSourceAction::NewEvent { event: _ } => true,

node/src/external_snark_worker/external_snark_worker_actions.rs

@@ -51,7 +51,7 @@ pub type ExternalSnarkWorkerActionWithMetaRef<'a> =
     redux::ActionWithMeta<&'a ExternalSnarkWorkerAction>;
 
 impl EnablingCondition<State> for ExternalSnarkWorkerAction {
-    fn is_enabled(&self, state: &State) -> bool {
+    fn is_enabled(&self, state: &State, _time: redux::Timestamp) -> bool {
         match self {
             ExternalSnarkWorkerAction::Start => {
                 state.config.snarker.is_some()

node/src/p2p/channels/best_tip/p2p_channels_best_tip_actions.rs

@@ -1,7 +1,7 @@
 use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pChannelsBestTipAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/channels/p2p_channels_actions.rs

@@ -1,7 +1,7 @@
 use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pChannelsMessageReceivedAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

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

@@ -1,13 +1,7 @@
 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),
-        }
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/channels/snark/p2p_channels_snark_actions.rs

@@ -1,7 +1,7 @@
 use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pChannelsSnarkAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/channels/snark_job_commitment/p2p_channels_snark_job_commitment_actions.rs

@@ -1,7 +1,7 @@
 use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pChannelsSnarkJobCommitmentAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

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

@@ -1,16 +1,7 @@
 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),
-        }
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

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),
-        }
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/disconnection/p2p_disconnection_actions.rs

@@ -1,7 +1,7 @@
 use super::*;
 
 impl redux::EnablingCondition<crate::State> for P2pDisconnectionAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/discovery/mod.rs

@@ -1,7 +1,7 @@
 pub use ::p2p::discovery::*;
 
 impl redux::EnablingCondition<crate::State> for P2pDiscoveryAction {
-    fn is_enabled(&self, state: &crate::State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/listen/p2p_listen_actions.rs

@@ -4,7 +4,7 @@ use redux::EnablingCondition;
 use crate::State;
 
 impl EnablingCondition<State> for P2pListenAction {
-    fn is_enabled(&self, state: &State) -> bool {
-        self.is_enabled(&state.p2p)
+    fn is_enabled(&self, state: &State, time: redux::Timestamp) -> bool {
+        self.is_enabled(&state.p2p, time)
     }
 }

node/src/p2p/mod.rs

@@ -32,7 +32,7 @@ where
 
     fn dispatch<A>(&mut self, action: A) -> bool
     where
-        A: Into<P2pAction> + redux::EnablingCondition<crate::State>,
+        A: Into<P2pAction> + redux::EnablingCondition<P2pState>,
     {
         crate::Store::sub_dispatch(self, action)
     }