Add a test for lots of nodes connecting at the same time (#6247)

* Add a test for lots of nodes connecting at the same time

* Do small change

substrate/client/network/src/service/tests.rs

@@ -272,3 +272,73 @@ fn notifications_state_consistent() {
 		}
 	});
 }
+
+#[test]
+fn lots_of_incoming_peers_works() {
+	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
+
+	let (main_node, _) = build_test_full_node(config::NetworkConfiguration {
+		notifications_protocols: vec![(ENGINE_ID, From::from(&b"/foo"[..]))],
+		listen_addresses: vec![listen_addr.clone()],
+		in_peers: u32::max_value(),
+		transport: config::TransportConfig::MemoryOnly,
+		.. config::NetworkConfiguration::new_local()
+	});
+
+	let main_node_peer_id = main_node.local_peer_id().clone();
+
+	// We spawn background tasks and push them in this `Vec`. They will all be waited upon before
+	// this test ends.
+	let mut background_tasks_to_wait = Vec::new();
+
+	for _ in 0..256 {
+		let main_node_peer_id = main_node_peer_id.clone();
+
+		let (_dialing_node, event_stream) = build_test_full_node(config::NetworkConfiguration {
+			notifications_protocols: vec![(ENGINE_ID, From::from(&b"/foo"[..]))],
+			listen_addresses: vec![],
+			reserved_nodes: vec![config::MultiaddrWithPeerId {
+				multiaddr: listen_addr.clone(),
+				peer_id: main_node_peer_id.clone(),
+			}],
+			transport: config::TransportConfig::MemoryOnly,
+			.. config::NetworkConfiguration::new_local()
+		});
+
+		background_tasks_to_wait.push(async_std::task::spawn(async move {
+			// Create a dummy timer that will "never" fire, and that will be overwritten when we
+			// actually need the timer. Using an Option would be technically cleaner, but it would
+			// make the code below way more complicated.
+			let mut timer = futures_timer::Delay::new(Duration::from_secs(3600 * 24 * 7)).fuse();
+
+			let mut event_stream = event_stream.fuse();
+			loop {
+				futures::select! {
+					_ = timer => {
+						// Test succeeds when timer fires.
+						return;
+					}
+					ev = event_stream.next() => {
+						match ev.unwrap() {
+							Event::NotificationStreamOpened { remote, .. } => {
+								assert_eq!(remote, main_node_peer_id);
+								// Test succeeds after 5 seconds. This timer is here in order to
+								// detect a potential problem after opening.
+								timer = futures_timer::Delay::new(Duration::from_secs(5)).fuse();
+							}
+							Event::NotificationStreamClosed { .. } => {
+								// Test failed.
+								panic!();
+							}
+							_ => {}
+						}
+					}
+				}
+			}
+		}));
+	}
+
+	futures::executor::block_on(async move {
+		future::join_all(background_tasks_to_wait).await
+	});
+}