fix(testing): remove some non-determinism from p2p

re: #188

Author: binier

node/testing/src/cluster/mod.rs

@@ -127,18 +127,17 @@ impl Cluster {
 
     pub fn add_rust_node(&mut self, testing_config: RustNodeTestingConfig) -> ClusterNodeId {
         let node_config = testing_config.clone();
-        let node_i = self.nodes.len();
+        let node_id = ClusterNodeId::new_unchecked(self.nodes.len());
         let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
         let secret_key = P2pSecretKey::from_bytes(match testing_config.peer_id {
             TestPeerId::Derived => {
                 let mut bytes = [0; 32];
                 let bytes_len = bytes.len();
-                let i_bytes = node_i.to_be_bytes();
+                let i_bytes = node_id.index().to_be_bytes();
                 let i = bytes_len - i_bytes.len();
                 bytes[i..bytes_len].copy_from_slice(&i_bytes);
                 bytes
             }
-            TestPeerId::Random => rand::random(),
             TestPeerId::Bytes(bytes) => bytes,
         });
         let pub_key = secret_key.public_key();
@@ -274,7 +273,7 @@ impl Cluster {
         if let Some(producer_key) = block_producer_sec_key {
             real_service.block_producer_start(producer_key.into());
         }
-        let mut service = NodeTestingService::new(real_service, shutdown_rx);
+        let mut service = NodeTestingService::new(real_service, node_id, shutdown_rx);
         if self.config.all_rust_to_rust_use_webrtc() {
             service.set_rust_to_rust_use_webrtc();
         }
@@ -314,7 +313,7 @@ impl Cluster {
         let node = Node::new(node_config, store);
 
         self.nodes.push(node);
-        ClusterNodeId::new_unchecked(node_i)
+        node_id
     }
 
     pub fn add_ocaml_node(&mut self, testing_config: OcamlNodeTestingConfig) -> ClusterOcamlNodeId {

node/testing/src/node/rust/config.rs

@@ -13,7 +13,6 @@ pub enum TestPeerId {
     /// runs (e.g. peer_id might be blacklisted).
     #[default]
     Derived,
-    Random,
     Bytes([u8; 32]),
 }
 
@@ -67,11 +66,6 @@ impl RustNodeTestingConfig {
         self
     }
 
-    pub fn with_random_peer_id(mut self) -> Self {
-        self.peer_id = TestPeerId::Random;
-        self
-    }
-
     pub fn with_peer_id(mut self, bytes: [u8; 32]) -> Self {
         self.peer_id = TestPeerId::Bytes(bytes);
         self

node/testing/src/scenarios/solo_node/basic_connectivity_accept_incoming.rs

@@ -5,6 +5,7 @@ use node::{
     event_source::Event,
     p2p::{connection::outgoing::P2pConnectionOutgoingInitOpts, P2pEvent, PeerId},
 };
+use rand::Rng;
 
 use crate::{
     node::{DaemonJson, OcamlNodeTestingConfig, RustNodeTestingConfig},
@@ -48,7 +49,7 @@ impl SoloNodeBasicConnectivityAcceptIncoming {
             .ask_initial_peers_interval(Duration::from_secs(3600))
             .max_peers(MAX_PEERS_PER_NODE)
             .initial_peers(initial_peers)
-            .with_random_peer_id();
+            .with_peer_id(rand::thread_rng().gen());
 
         let node_id = runner.add_rust_node(config);
         let node_addr = runner.node(node_id).unwrap().dial_addr();

node/testing/src/service/mod.rs

@@ -36,6 +36,8 @@ use node::{ActionWithMeta, State};
 use openmina_node_native::NodeService;
 use redux::Instant;
 
+use crate::cluster::ClusterNodeId;
+
 pub type DynEffects = Box<dyn FnMut(&State, &NodeTestingService, &ActionWithMeta) + Send>;
 
 #[derive(Hash, Ord, PartialOrd, Eq, PartialEq)]
@@ -49,6 +51,7 @@ pub type PendingEventId = RequestId<PendingEventIdType>;
 
 pub struct NodeTestingService {
     real: NodeService,
+    id: ClusterNodeId,
     // Use webrtc p2p between Rust nodes.
     rust_to_rust_use_webrtc: bool,
     monotonic_time: Instant,
@@ -60,9 +63,10 @@ pub struct NodeTestingService {
 }
 
 impl NodeTestingService {
-    pub fn new(real: NodeService, _shutdown: mpsc::Receiver<()>) -> Self {
+    pub fn new(real: NodeService, id: ClusterNodeId, _shutdown: mpsc::Receiver<()>) -> Self {
         Self {
             real,
+            id,
             rust_to_rust_use_webrtc: false,
             monotonic_time: Instant::now(),
             pending_events: PendingRequests::new(),
@@ -201,7 +205,23 @@ impl P2pServiceWebrtcWithLibp2p for NodeTestingService {
     }
 
     fn find_random_peer(&mut self) {
-        self.real.find_random_peer()
+        use node::p2p::identity::SecretKey as P2pSecretKey;
+        use node::p2p::service_impl::libp2p::Cmd;
+
+        let secret_key = P2pSecretKey::from_bytes({
+            let mut bytes = [1; 32];
+            let bytes_len = bytes.len();
+            let i_bytes = self.id.index().to_be_bytes();
+            let i = bytes_len - i_bytes.len();
+            bytes[i..bytes_len].copy_from_slice(&i_bytes);
+            bytes
+        });
+        let peer_id = secret_key.public_key().peer_id();
+
+        self.libp2p()
+            .cmd_sender()
+            .send(Cmd::FindNode(peer_id.into()))
+            .unwrap_or_default();
     }
 
     fn start_discovery(&mut self, peers: Vec<P2pConnectionOutgoingInitOpts>) {

p2p/src/p2p_event.rs

@@ -81,35 +81,45 @@ impl fmt::Display for P2pEvent {
             Self::Listen(v) => v.fmt(f),
             Self::Channel(v) => v.fmt(f),
             #[cfg(not(target_arch = "wasm32"))]
-            Self::Libp2pIdentify(peer_id, addr) => {
-                write!(f, "{peer_id} {addr}")
+            Self::Libp2pIdentify(peer_id, _) => {
+                write!(f, "Libp2pIdentify, {peer_id}")
             }
             Self::Discovery(v) => v.fmt(f),
         }
     }
 }
 
+fn maddr_ip(addr: &libp2p::Multiaddr) -> String {
+    use libp2p::multiaddr::Protocol;
+    addr.iter()
+        .find_map(|p| match p {
+            Protocol::Ip4(_)
+            | Protocol::Ip6(_)
+            | Protocol::Dns(_)
+            | Protocol::Dns4(_)
+            | Protocol::Dns6(_) => Some(p.to_string()),
+            _ => None,
+        })
+        .unwrap_or(String::new())
+}
+
 impl fmt::Display for P2pListenEvent {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "Listen, ")?;
         match self {
-            P2pListenEvent::NewListenAddr { listener_id, addr } => {
-                write!(f, "NewListenAddr, {listener_id}, {addr}")
+            P2pListenEvent::NewListenAddr { addr, .. } => {
+                write!(f, "NewListenAddr, {}", maddr_ip(addr))
             }
-            P2pListenEvent::ExpiredListenAddr { listener_id, addr } => {
-                write!(f, "ExpiredListenAddr, {listener_id}, {addr}")
+            P2pListenEvent::ExpiredListenAddr { addr, .. } => {
+                write!(f, "ExpiredListenAddr, {}", maddr_ip(addr))
             }
-            P2pListenEvent::ListenerError { listener_id, error } => {
-                write!(f, "ListenerError, {listener_id}, {error}")
+            P2pListenEvent::ListenerError { error, .. } => {
+                write!(f, "ListenerError, {error}")
             }
             P2pListenEvent::ListenerClosed {
-                listener_id,
-                error: Some(error),
-            } => write!(f, "ListenerClosed, {listener_id}, {error}"),
-            P2pListenEvent::ListenerClosed {
-                listener_id,
-                error: None,
-            } => write!(f, "ListenerClosed, {listener_id}"),
+                error: Some(error), ..
+            } => write!(f, "ListenerClosed, {error}"),
+            P2pListenEvent::ListenerClosed { error: None, .. } => write!(f, "ListenerClosed"),
         }
     }
 }