/* This file is part 10f DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2024 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
// cargo +nightly test --release --features=net --lib p2p -- --include-ignored
use std::{collections::HashSet, sync::Arc};
use log::{info, warn};
use rand::{prelude::SliceRandom, rngs::ThreadRng, Rng};
use smol::{channel, future, Executor};
use url::Url;
use crate::{
net::{hosts::HostColor, P2p, Settings},
system::sleep,
};
// Number of nodes to spawn and number of peers each node connects to
const N_NODES: usize = 5;
//const N_CONNS: usize = 5;
const SEED: &str = "tcp://127.0.0.1:51505";
fn init_logger() {
let mut cfg = simplelog::ConfigBuilder::new();
cfg.add_filter_ignore("sled".to_string());
cfg.add_filter_ignore("net::protocol_ping".to_string());
cfg.add_filter_ignore("net::channel::subscribe_stop()".to_string());
cfg.add_filter_ignore("net::hosts".to_string());
cfg.add_filter_ignore("net::session".to_string());
cfg.add_filter_ignore("net::outbound_session".to_string());
cfg.add_filter_ignore("net::inbound_session".to_string());
cfg.add_filter_ignore("net::message_subscriber".to_string());
cfg.add_filter_ignore("net::protocol_address".to_string());
cfg.add_filter_ignore("net::protocol_version".to_string());
cfg.add_filter_ignore("net::protocol_registry".to_string());
cfg.add_filter_ignore("net::protocol_jobs_manager".to_string());
cfg.add_filter_ignore("net::channel::send()".to_string());
cfg.add_filter_ignore("net::channel::start()".to_string());
cfg.add_filter_ignore("net::channel::handle_stop()".to_string());
cfg.add_filter_ignore("net::channel::subscribe_msg()".to_string());
cfg.add_filter_ignore("net::channel::main_receive_loop()".to_string());
cfg.add_filter_ignore("net::tcp".to_string());
// We check this error so we can execute same file tests in parallel,
// otherwise second one fails to init logger here.
if simplelog::TermLogger::init(
simplelog::LevelFilter::Info,
//simplelog::LevelFilter::Debug,
//simplelog::LevelFilter::Trace,
cfg.build(),
simplelog::TerminalMode::Mixed,
simplelog::ColorChoice::Auto,
)
.is_err()
{
warn!(target: "test_harness", "Logger already initialized");
}
}
async fn spawn_node(
inbound_addrs: Vec,
external_addrs: Vec,
peers: Vec,
seeds: Vec,
node_id: String,
ex: Arc>,
) -> Arc {
let settings = Settings {
localnet: true,
inbound_addrs,
external_addrs,
outbound_connections: 2,
outbound_peer_discovery_cooloff_time: 2,
//outbound_connect_timeout: 2,
inbound_connections: usize::MAX,
greylist_refinery_interval: 15,
peers,
seeds,
node_id,
allowed_transports: vec!["tcp".to_string()],
..Default::default()
};
P2p::new(settings, ex.clone()).await
}
async fn spawn_seed_session(starting_port: usize, ex: Arc>) -> Vec> {
let mut p2p_instances = vec![];
let seed_addr = Url::parse(SEED).unwrap();
info!("========================================================");
info!("Initializing outbound nodes...");
info!("========================================================");
for i in 0..N_NODES {
let p2p = spawn_node(
vec![Url::parse(&format!("tcp://127.0.0.1:{}", starting_port + i)).unwrap()],
vec![Url::parse(&format!("tcp://127.0.0.1:{}", starting_port + i)).unwrap()],
vec![],
vec![seed_addr.clone()],
(starting_port + i).to_string(),
ex.clone(),
)
.await;
p2p_instances.push(p2p);
}
// Start the P2P network
for p2p in p2p_instances.iter() {
info!("========================================================");
info!("Starting node={}", p2p.settings().external_addrs[0]);
info!("========================================================");
p2p.clone().start().await.unwrap();
}
p2p_instances
}
/*async fn spawn_manual_session(
peer_indexes: &[usize],
starting_port: usize,
rng: &mut ThreadRng,
ex: Arc>,
) -> Vec> {
let mut p2p_instances = vec![];
// Initialize the nodes
for i in 0..N_NODES {
// Everyone will connect to N_CONNS random peers.
let mut peer_indexes_copy = peer_indexes.to_owned();
peer_indexes_copy.remove(i);
let peer_indexes_to_connect: Vec<_> =
peer_indexes_copy.choose_multiple(rng, N_CONNS).collect();
let mut peers = vec![];
for peer_index in peer_indexes_to_connect {
let port = starting_port + peer_index;
peers.push(Url::parse(&format!("tcp://127.0.0.1:{}", port)).unwrap());
}
let p2p = spawn_node(
vec![Url::parse(&format!("tcp://127.0.0.1:{}", starting_port + i)).unwrap()],
vec![Url::parse(&format!("tcp://127.0.0.1:{}", starting_port + i)).unwrap()],
vec![],
peers,
(starting_port + i).to_string(),
ex.clone(),
)
.await;
p2p_instances.push(p2p);
}
// Start the P2P network
for p2p in p2p_instances.iter() {
p2p.clone().start().await.unwrap();
}
info!("Waiting 5s until all peers connect");
sleep(5).await;
p2p_instances
}*/
/*async fn assert_hostlist_not_empty(
p2p_instances: &Vec>,
rng: &mut ThreadRng,
color: HostColor,
) {
let random_node = p2p_instances.choose(rng).unwrap();
assert!(!random_node.hosts().container.is_empty(color).await);
}*/
/*async fn assert_entry_exists(
p2p_instances: &Vec>,
rng: &mut ThreadRng,
color: HostColor,
entry: &Url,
) {
let mut urls = HashSet::new();
let random_node = p2p_instances.choose(rng).unwrap();
let external_addr = &random_node.settings().external_addrs[0];
info!("Checking {} entry exists on {:?} list node={}", entry, color, external_addr);
assert!(random_node.hosts().container.contains(color as usize, entry).await);
}*/
async fn get_random_gold_host(p2p_instances: &[Arc], index: usize) -> ((Url, u64), usize) {
let random_node = &p2p_instances[index];
let external_addr = &random_node.settings().external_addrs[0];
info!("========================================================");
info!("Getting gold addr from node={}", external_addr);
info!("========================================================");
random_node.hosts().container.fetch_random(HostColor::Gold).await
}
async fn check_random_hostlist(p2p_instances: &Vec>, rng: &mut ThreadRng) {
let mut urls = HashSet::new();
let random_node = p2p_instances.choose(rng).unwrap();
let external_addr = &random_node.settings().external_addrs[0];
info!("========================================================");
info!("Checking node={}", external_addr);
info!("========================================================");
let greylist = random_node.hosts().container.fetch_all(HostColor::Grey).await;
let whitelist = random_node.hosts().container.fetch_all(HostColor::White).await;
let goldlist = random_node.hosts().container.fetch_all(HostColor::Gold).await;
for (url, _) in greylist {
assert!(urls.insert(url));
}
for (url, _) in whitelist {
assert!(urls.insert(url));
}
for (url, _) in goldlist {
assert!(urls.insert(url));
}
assert!(!urls.is_empty());
}
async fn kill_node(p2p_instances: &Vec>, node: Url) {
for p2p in p2p_instances {
if p2p.settings().external_addrs[0] == node {
info!("========================================================");
info!("Shutting down node: {}", p2p.settings().external_addrs[0]);
info!("========================================================");
p2p.stop().await;
}
}
}
macro_rules! test_body {
($real_call:ident) => {
init_logger();
let ex = Arc::new(Executor::new());
let ex_ = ex.clone();
let (signal, shutdown) = channel::unbounded::<()>();
// Run a thread for each node.
easy_parallel::Parallel::new()
.each(0..N_NODES, |_| future::block_on(ex.run(shutdown.recv())))
.finish(|| {
future::block_on(async {
$real_call(ex_).await;
drop(signal);
})
});
};
}
#[test]
fn p2p_test() {
test_body!(p2p_test_real);
}
async fn p2p_test_real(ex: Arc>) {
let mut rng = rand::thread_rng();
// ============================================================
// 1. Create a new seed node.
// ============================================================
//let peer_indexes: Vec = (0..N_NODES).collect();
let seed_addr = Url::parse(SEED).unwrap();
let settings = Settings {
localnet: true,
inbound_addrs: vec![seed_addr.clone()],
outbound_connections: 0,
inbound_connections: usize::MAX,
seeds: vec![],
peers: vec![],
allowed_transports: vec!["tcp".to_string()],
greylist_refinery_interval: 12,
node_id: "seed".to_string(),
..Default::default()
};
let seed = P2p::new(settings, ex.clone()).await;
info!("========================================================");
info!("Starting seed node on {}", SEED);
info!("========================================================");
seed.clone().start().await.unwrap();
// ============================================================
// 2. Spawn outbound nodes that will connect to the seed node.
// ============================================================
let p2p_instances = spawn_seed_session(43200, ex.clone()).await;
info!("========================================================");
info!("Waiting 10s for all peers to reach the seed node");
info!("========================================================");
sleep(10).await;
// ===========================================================
// 3. Assert that all nodes have shared their external addr
// with the seed node.
// ===========================================================
let greylist = seed.hosts().container.fetch_all(HostColor::Grey).await;
assert!(greylist.len() == N_NODES);
info!("========================================================");
info!("Seedsync session successful!");
info!("========================================================");
info!("========================================================");
info!("Waiting 5s for seed node refinery to kick in...");
info!("========================================================");
sleep(5).await;
// ===========================================================
// 4. Assert that seed node has at least one whitelist entry,
// indicating that the refinery process is happening correctly.
// ===========================================================
assert!(!seed.hosts().container.is_empty(HostColor::White).await);
info!("========================================================");
info!("Seed node refinery operating successfully!");
info!("========================================================");
info!("========================================================");
info!("Waiting 5s for peers to propagate...");
info!("========================================================");
sleep(5).await;
// ===========================================================
// 5. Select a random peer and ensure that its hostlist is not
// empty. This ensures the seed node is sharing whitelisted
// nodes around the network.
// ===========================================================
check_random_hostlist(&p2p_instances, &mut rng).await;
info!("========================================================");
info!("Peer successfully received addrs!");
info!("========================================================");
// ===========================================================
// 6. Select a random gold peer from one of the nodes and kill
// it.
// ===========================================================
info!("========================================================");
info!("Selecting a random gold entry...");
info!("========================================================");
let random_node_index = rand::thread_rng().gen_range(0..p2p_instances.len());
let ((addr, _), _) = get_random_gold_host(&p2p_instances, random_node_index).await;
kill_node(&p2p_instances, addr.clone()).await;
info!("========================================================");
info!("Waiting for greylist downgrade sequence to occur...");
info!("========================================================");
// ===========================================================
// 7. Verify the peer has been removed from the Gold list.
// ===========================================================
p2p_instances[random_node_index]
.hosts()
.container
.contains(HostColor::Grey as usize, &addr)
.await;
info!("========================================================");
info!("Greylist downgrade occured successfully!");
info!("========================================================");
// ===========================================================
// 8. Stop the P2P network
// ===========================================================
for p2p in p2p_instances.iter() {
p2p.clone().stop().await;
}
}