Merge pull request #611 from openmina/feat/p2p/meshsub

vlad9486 · web-flow · commit dd3bca391a55 · 2024-08-30T15:05:32.000+02:00
Pubsub Bugs
diff --git a/node/testing/src/scenarios/multi_node/pubsub_advanced.rs b/node/testing/src/scenarios/multi_node/pubsub_advanced.rs
@@ -1,11 +1,12 @@
-use std::{iter, str::FromStr, time::Duration};
+use std::time::Duration;
 
-use node::{account::AccountSecretKey, BlockProducerConfig};
+use mina_p2p_messages::v2;
+use node::transition_frontier::genesis::{GenesisConfig, NonStakers};
 
 use crate::{
-    node::{RustNodeBlockProducerTestingConfig, RustNodeTestingConfig},
-    scenario::ListenerNode,
-    scenarios::ClusterRunner,
+    node::Recorder,
+    scenarios::{ClusterRunner, RunCfgAdvanceTime},
+    simulator::{Simulator, SimulatorConfig, SimulatorRunUntil},
 };
 
 /// Create and Sync up 50 nodes, one amoung them is block producer.
@@ -19,66 +20,31 @@ use crate::{
 pub struct MultiNodePubsubPropagateBlock;
 
 impl MultiNodePubsubPropagateBlock {
-    const WORKERS: usize = 4;
+    const WORKERS: usize = 20;
 
     pub async fn run(self, mut runner: ClusterRunner<'_>) {
-        // let seed_node = ListenerNode::Custom(
-        //     "/ip4/34.135.63.47/tcp/10001/p2p/12D3KooWLjs54xHzVmMmGYb7W5RVibqbwD1co7M2ZMfPgPm7iAag"
-        //         .parse()
-        //         .unwrap(),
-        // );
-
-        let seed_config = RustNodeTestingConfig::devnet_default()
-            .max_peers(100)
-            .ask_initial_peers_interval(Duration::from_secs(60 * 60));
-
-        let seed_node = ListenerNode::Rust(runner.add_rust_node(seed_config));
-
-        // for account B62qrztYfPinaKqpXaYGY6QJ3SSW2NNKs7SajBLF1iFNXW9BoALN2Aq
-        let sec_key =
-            AccountSecretKey::from_str("EKEEpMELfQkMbJDt2fB4cFXKwSf1x4t7YD4twREy5yuJ84HBZtF9")
-                .unwrap();
-
-        let producer_node = runner.add_rust_node(RustNodeTestingConfig {
-            initial_time: redux::Timestamp::ZERO,
-            genesis: node::config::DEVNET_CONFIG.clone(),
-            max_peers: 10,
-            ask_initial_peers_interval: Duration::from_secs(60 * 60),
-            initial_peers: vec![seed_node.clone()],
-            peer_id: Default::default(),
-            block_producer: Some(RustNodeBlockProducerTestingConfig {
-                config: BlockProducerConfig {
-                    pub_key: sec_key.public_key().into(),
-                    custom_coinbase_receiver: None,
-                    proposed_protocol_version: None,
-                },
-                sec_key,
-            }),
-            snark_worker: None,
-            timeouts: Default::default(),
-            libp2p_port: None,
-            recorder: Default::default(),
-        });
-
-        tokio::time::sleep(Duration::from_secs(2)).await;
-
-        eprintln!("Producer node connected");
-
-        let worker_config = RustNodeTestingConfig::devnet_default()
-            .initial_peers(vec![seed_node])
-            .max_peers(10)
-            .ask_initial_peers_interval(Duration::from_secs(60 * 60));
-        let workers = iter::repeat(worker_config)
-            .take(Self::WORKERS)
-            .map(|config| runner.add_rust_node(config))
-            .collect::<Vec<_>>();
-
-        runner
-            .run_until_nodes_synced(Duration::from_secs(10 * 60), &workers)
-            .await
-            .unwrap();
-
-        let _ = producer_node;
-        // TODO:
+        let initial_time = redux::Timestamp::global_now();
+        let mut constants = v2::PROTOCOL_CONSTANTS.clone();
+        constants.genesis_state_timestamp =
+            v2::BlockTimeTimeStableV1((u64::from(initial_time) / 1_000_000).into());
+        let genesis_cfg = GenesisConfig::Counts {
+            whales: 1,
+            fish: 0,
+            non_stakers: NonStakers::None,
+            constants,
+        };
+        let config = SimulatorConfig {
+            genesis: genesis_cfg.into(),
+            seed_nodes: 1,
+            normal_nodes: Self::WORKERS,
+            snark_workers: 1,
+            block_producers: 1,
+            advance_time: RunCfgAdvanceTime::Rand(1..=200),
+            run_until: SimulatorRunUntil::BlockchainLength(3),
+            run_until_timeout: Duration::from_secs(10 * 60),
+            recorder: Recorder::StateWithInputActions,
+        };
+        let mut simulator = Simulator::new(initial_time, config);
+        simulator.run(&mut runner).await;
     }
 }
diff --git a/node/testing/tests/multi_node.rs b/node/testing/tests/multi_node.rs
@@ -17,9 +17,9 @@ scenario_test!(
     MultiNodeBasicConnectivityInitialJoining
 );
 
-// #[cfg(feature = "p2p-libp2p")]
-// scenario_test!(
-//     propagate_block,
-//     openmina_node_testing::scenarios::multi_node::pubsub_advanced::MultiNodePubsubPropagateBlock,
-//     openmina_node_testing::scenarios::multi_node::pubsub_advanced::MultiNodePubsubPropagateBlock
-// );
+#[cfg(feature = "p2p-libp2p")]
+scenario_test!(
+    propagate_block,
+    openmina_node_testing::scenarios::multi_node::pubsub_advanced::MultiNodePubsubPropagateBlock,
+    openmina_node_testing::scenarios::multi_node::pubsub_advanced::MultiNodePubsubPropagateBlock
+);
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_effects.rs b/p2p/src/network/pubsub/p2p_network_pubsub_effects.rs
@@ -23,10 +23,13 @@ impl P2pNetworkPubsubAction {
         let state = &store.state().network.scheduler.broadcast_state;
         let config = &store.state().config;
 
+        // let this = config.identity_pub_key.peer_id();
+
         match self {
             P2pNetworkPubsubAction::NewStream {
                 peer_id, incoming, ..
             } => {
+                // println!("(pubsub) {this} new stream {peer_id} {incoming}");
                 if !incoming {
                     let subscrption = {
                         let msg = pb::Rpc {
@@ -65,7 +68,7 @@ impl P2pNetworkPubsubAction {
                         ihave: vec![],
                         iwant: vec![],
                         graft: vec![pb::ControlGraft {
-                            topic_id: Some(dbg!(topic_id.clone())),
+                            topic_id: Some(topic_id.clone()),
                         }],
                         prune: vec![],
                     }),
@@ -82,7 +85,7 @@ impl P2pNetworkPubsubAction {
                         iwant: vec![],
                         graft: vec![],
                         prune: vec![pb::ControlPrune {
-                            topic_id: Some(dbg!(topic_id.clone())),
+                            topic_id: Some(topic_id.clone()),
                             peers: vec![pb::PeerInfo {
                                 peer_id: None,
                                 signed_peer_record: None,
@@ -95,6 +98,7 @@ impl P2pNetworkPubsubAction {
                 store.dispatch(P2pNetworkPubsubAction::OutgoingMessage { msg, peer_id });
             }
             P2pNetworkPubsubAction::Broadcast { message } => {
+                // println!("(pubsub) {this} broadcast");
                 let mut buffer = vec![0; 8];
                 binprot::BinProtWrite::binprot_write(&message, &mut buffer).expect("msg");
                 let len = buffer.len() - 8;
@@ -117,6 +121,8 @@ impl P2pNetworkPubsubAction {
             }
             P2pNetworkPubsubAction::BroadcastSigned { .. } => broadcast(store),
             P2pNetworkPubsubAction::IncomingData { peer_id, .. } => {
+                // println!("(pubsub) {this} <- {peer_id}");
+
                 let incoming_block = state.incoming_block.as_ref().cloned();
                 let incoming_transactions = state.incoming_transactions.clone();
                 let incoming_snarks = state.incoming_snarks.clone();
@@ -162,6 +168,16 @@ impl P2pNetworkPubsubAction {
             }
             P2pNetworkPubsubAction::OutgoingMessage { msg, peer_id } => {
                 if !message_is_empty(&msg) {
+                    // println!(
+                    //     "(pubsub) {this} -> {peer_id}, {:?}, {:?}, {}",
+                    //     msg.subscriptions,
+                    //     msg.control,
+                    //     msg.publish.len()
+                    // );
+                    // for ele in &msg.publish {
+                    //     let id = super::p2p_network_pubsub_state::compute_message_id(ele);
+                    //     println!("{}", std::str::from_utf8(&id).unwrap());
+                    // }
                     let mut data = vec![];
                     if prost::Message::encode_length_delimited(&msg, &mut data).is_ok() {
                         store.dispatch(P2pNetworkPubsubAction::OutgoingData {
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs b/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs
@@ -37,6 +37,9 @@ impl P2pNetworkPubsubState {
                 });
                 state.protocol = *protocol;
                 state.addr = *addr;
+
+                self.topics
+                    .insert(super::TOPIC.to_owned(), Default::default());
             }
             P2pNetworkPubsubAction::NewStream {
                 incoming: false,
@@ -83,6 +86,13 @@ impl P2pNetworkPubsubState {
                 match <pb::Rpc as prost::Message>::decode_length_delimited(slice) {
                     Ok(v) => {
                         state.buffer.clear();
+                        // println!(
+                        //     "(pubsub) this <- {peer_id}, {:?}, {:?}, {}",
+                        //     v.subscriptions,
+                        //     v.control,
+                        //     v.publish.len()
+                        // );
+
                         for subscription in v.subscriptions {
                             let topic_id = subscription.topic_id().to_owned();
                             let topic = self.topics.entry(topic_id).or_default();
@@ -169,7 +179,7 @@ impl P2pNetworkPubsubState {
                             for graft in &control.graft {
                                 if let Some(mesh_state) = self
                                     .topics
-                                    .get_mut(dbg!(graft.topic_id()))
+                                    .get_mut(graft.topic_id())
                                     .and_then(|m| m.get_mut(peer_id))
                                 {
                                     mesh_state.mesh = P2pNetworkPubsubClientMeshAddingState::Added;
diff --git a/p2p/src/p2p_config.rs b/p2p/src/p2p_config.rs
@@ -57,7 +57,7 @@ impl Default for P2pMeshsubConfig {
             outbound_degree_desired: 6,
             outbound_degree_low: 4,
             outbound_degree_high: 12,
-            mcache_len: 5,
+            mcache_len: 256,
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ impl Default for P2pMeshsubConfig {`
`57`	`57`	`outbound_degree_desired: 6,`
`58`	`58`	`outbound_degree_low: 4,`
`59`	`59`	`outbound_degree_high: 12,`
`60`		`- mcache_len: 5,`
	`60`	`+ mcache_len: 256,`
`61`	`61`	`}`
`62`	`62`	`}`
`63`	`63`	`}`