gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-23 23:04:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Adds simulator for syncing (#758)

Browse Source 
* Add CLI for beacon_chain_sim

* Rename beacon-chain-sim to simulator

* Fix simulator workflow

* Push Cargo.lock

* WIP syncing simulator

* Add cli args

* Remove eth1 stuff and deposits

* Add syncing strategy simulations

* Successful one node sync

* Clean up

* Rename to avoid confusion

* add command line args

* fix cargo fmt issues

* Add additional syncing strategies

* Run all syncing strategies one after other; add comments

* Improve cli argument parsing

* Change `end_after_checks` default to true

* Small modifications to syncing-sim

* Add `strategy` cli argument

* Documented defaults in cli help

Co-authored-by: mkinney <mike.kinney@gmail.com>
Co-authored-by: Age Manning <Age@AgeManning.com>
This commit is contained in:
Pawan Dhananjay
2020-01-23 12:37:39 +05:30
committed by Age Manning
parent 8c96739cab
commit 89f05e4a4f
9 changed files with 461 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
tests/simulator/src/checks.rs Normal file
View File

@@ -0,0 +1,136 @@
use crate::local_network::LocalNetwork;
use futures::{stream, Future, IntoFuture, Stream};
use std::time::{Duration, Instant};
use tokio::timer::Delay;
use types::{Epoch, EthSpec, Slot, Unsigned};
/// Checks that all of the validators have on-boarded by the start of the second eth1 voting
/// period.
pub fn verify_initial_validator_count<E: EthSpec>(
network: LocalNetwork<E>,
slot_duration: Duration,
initial_validator_count: usize,
) -> impl Future<Item = (), Error = String> {
slot_delay(Slot::new(1), slot_duration)
.and_then(move |()| verify_validator_count(network, initial_validator_count))
}
/// Checks that all of the validators have on-boarded by the start of the second eth1 voting
/// period.
pub fn verify_validator_onboarding<E: EthSpec>(
network: LocalNetwork<E>,
slot_duration: Duration,
expected_validator_count: usize,
) -> impl Future<Item = (), Error = String> {
slot_delay(
Slot::new(E::SlotsPerEth1VotingPeriod::to_u64()),
slot_duration,
)
.and_then(move |()| verify_validator_count(network, expected_validator_count))
}
/// Checks that the chain has made the first possible finalization.
///
/// Intended to be run as soon as chain starts.
pub fn verify_first_finalization<E: EthSpec>(
network: LocalNetwork<E>,
slot_duration: Duration,
) -> impl Future<Item = (), Error = String> {
epoch_delay(Epoch::new(4), slot_duration, E::slots_per_epoch())
.and_then(|()| verify_all_finalized_at(network, Epoch::new(2)))
}
/// Delays for `epochs`, plus half a slot extra.
pub fn epoch_delay(
epochs: Epoch,
slot_duration: Duration,
slots_per_epoch: u64,
) -> impl Future<Item = (), Error = String> {
let duration = slot_duration * (epochs.as_u64() * slots_per_epoch) as u32 + slot_duration / 2;
Delay::new(Instant::now() + duration).map_err(|e| format!("Epoch delay failed: {:?}", e))
}
/// Delays for `slots`, plus half a slot extra.
fn slot_delay(slots: Slot, slot_duration: Duration) -> impl Future<Item = (), Error = String> {
let duration = slot_duration * slots.as_u64() as u32 + slot_duration / 2;
Delay::new(Instant::now() + duration).map_err(|e| format!("Epoch delay failed: {:?}", e))
}
/// Verifies that all beacon nodes in the given network have a head state that has a finalized
/// epoch of `epoch`.
pub fn verify_all_finalized_at<E: EthSpec>(
network: LocalNetwork<E>,
epoch: Epoch,
) -> impl Future<Item = (), Error = String> {
network
.remote_nodes()
.into_future()
.and_then(|remote_nodes| {
stream::unfold(remote_nodes.into_iter(), |mut iter| {
iter.next().map(|remote_node| {
remote_node
.http
.beacon()
.get_head()
.map(|head| head.finalized_slot.epoch(E::slots_per_epoch()))
.map(|epoch| (epoch, iter))
.map_err(|e| format!("Get head via http failed: {:?}", e))
})
})
.collect()
})
.and_then(move |epochs| {
if epochs.iter().any(|node_epoch| *node_epoch != epoch) {
Err(format!(
"Nodes are not finalized at epoch {}. Finalized epochs: {:?}",
epoch, epochs
))
} else {
Ok(())
}
})
}
/// Verifies that all beacon nodes in the given `network` have a head state that contains
/// `expected_count` validators.
fn verify_validator_count<E: EthSpec>(
network: LocalNetwork<E>,
expected_count: usize,
) -> impl Future<Item = (), Error = String> {
network
.remote_nodes()
.into_future()
.and_then(|remote_nodes| {
stream::unfold(remote_nodes.into_iter(), |mut iter| {
iter.next().map(|remote_node| {
let beacon = remote_node.http.beacon();
beacon
.get_head()
.map_err(|e| format!("Get head via http failed: {:?}", e))
.and_then(move |head| {
beacon
.get_state_by_root(head.state_root)
.map(|(state, _root)| state)
.map_err(|e| format!("Get state root via http failed: {:?}", e))
})
.map(|state| (state.validators.len(), iter))
})
})
.collect()
})
.and_then(move |validator_counts| {
if validator_counts
.iter()
.any(|count| *count != expected_count)
{
Err(format!(
"Nodes do not all have {} validators in their state. Validator counts: {:?}",
expected_count, validator_counts
))
} else {
Ok(())
}
})
}

73
tests/simulator/src/cli.rs Normal file
View File

@@ -0,0 +1,73 @@
use clap::{App, Arg, SubCommand};
pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
App::new("simulator")
.version(crate_version!())
.author("Sigma Prime <contact@sigmaprime.io>")
.about("Options for interacting with simulator")
.subcommand(
SubCommand::with_name("beacon-chain-sim")
.about(
"Lighthouse Beacon Chain Simulator creates `n` beacon node and validator clients, \
each with `v` validators. A deposit contract is deployed at the start of the \
simulation using a local `ganache-cli` instance (you must have `ganache-cli` \
installed and avaliable on your path). All beacon nodes independently listen \
for genesis from the deposit contract, then start operating. \
As the simulation runs, there are checks made to ensure that all components \
are running correctly. If any of these checks fail, the simulation will \
exit immediately.",
)
.arg(Arg::with_name("nodes")
.short("n")
.long("nodes")
.takes_value(true)
.help("Number of beacon nodes (default 4)"))
.arg(Arg::with_name("validators_per_node")
.short("v")
.long("validators_per_node")
.takes_value(true)
.help("Number of validators (default 20)"))
.arg(Arg::with_name("speed_up_factor")
.short("s")
.long("speed_up_factor")
.takes_value(true)
.help("Speed up factor (default 4)"))
.arg(Arg::with_name("end_after_checks")
.short("e")
.long("end_after_checks")
.takes_value(false)
.help("End after checks (default true)"))
)
.subcommand(
SubCommand::with_name("syncing-sim")
.about("Run the syncing simulation")
.arg(
Arg::with_name("speedup")
.short("s")
.long("speedup")
.takes_value(true)
.help("Speed up factor for eth1 blocks and slot production (default 15)"),
)
.arg(
Arg::with_name("initial_delay")
.short("i")
.long("initial_delay")
.takes_value(true)
.help("Epoch delay for new beacon node to start syncing (default 50)"),
)
.arg(
Arg::with_name("sync_delay")
.long("sync_delay")
.takes_value(true)
.help("Epoch delay for newly added beacon nodes get synced (default 10)"),
)
.arg(
Arg::with_name("strategy")
.long("strategy")
.takes_value(true)
.possible_values(&["one-node", "two-nodes", "mixed", "all"])
.help("Sync strategy to run. (default all)"),
),
)
}

169
tests/simulator/src/local_network.rs Normal file
View File

@@ -0,0 +1,169 @@
use futures::{Future, IntoFuture};
use node_test_rig::{
environment::RuntimeContext, ClientConfig, LocalBeaconNode, LocalValidatorClient,
RemoteBeaconNode, ValidatorConfig,
};
use parking_lot::RwLock;
use std::ops::Deref;
use std::sync::Arc;
use types::{Epoch, EthSpec};
const BOOTNODE_PORT: u16 = 42424;
/// Helper struct to reduce `Arc` usage.
pub struct Inner<E: EthSpec> {
context: RuntimeContext<E>,
beacon_nodes: RwLock<Vec<LocalBeaconNode<E>>>,
validator_clients: RwLock<Vec<LocalValidatorClient<E>>>,
}
/// Represents a set of interconnected `LocalBeaconNode` and `LocalValidatorClient`.
///
/// Provides functions to allow adding new beacon nodes and validators.
pub struct LocalNetwork<E: EthSpec> {
inner: Arc<Inner<E>>,
}
impl<E: EthSpec> Clone for LocalNetwork<E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<E: EthSpec> Deref for LocalNetwork<E> {
type Target = Inner<E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<E: EthSpec> LocalNetwork<E> {
/// Creates a new network with a single `BeaconNode`.
pub fn new(
context: RuntimeContext<E>,
mut beacon_config: ClientConfig,
) -> impl Future<Item = Self, Error = String> {
// Fix bootnode ports
beacon_config.network.discovery_port = BOOTNODE_PORT;
beacon_config.network.libp2p_port = BOOTNODE_PORT;
LocalBeaconNode::production(context.service_context("boot_node".into()), beacon_config).map(
|beacon_node| Self {
inner: Arc::new(Inner {
context,
beacon_nodes: RwLock::new(vec![beacon_node]),
validator_clients: RwLock::new(vec![]),
}),
},
)
}
/// Returns the number of beacon nodes in the network.
///
/// Note: does not count nodes that are external to this `LocalNetwork` that may have connected
/// (e.g., another Lighthouse process on the same machine.)
pub fn beacon_node_count(&self) -> usize {
self.beacon_nodes.read().len()
}
/// Returns the number of validator clients in the network.
///
/// Note: does not count nodes that are external to this `LocalNetwork` that may have connected
/// (e.g., another Lighthouse process on the same machine.)
pub fn validator_client_count(&self) -> usize {
self.validator_clients.read().len()
}
/// Adds a beacon node to the network, connecting to the 0'th beacon node via ENR.
pub fn add_beacon_node(
&self,
mut beacon_config: ClientConfig,
) -> impl Future<Item = (), Error = String> {
let self_1 = self.clone();
println!("Adding beacon node..");
self.beacon_nodes
.read()
.first()
.map(|boot_node| {
beacon_config.network.boot_nodes.push(
boot_node
.client
.enr()
.expect("bootnode must have a network"),
);
})
.expect("should have atleast one node");
let index = self.beacon_nodes.read().len();
LocalBeaconNode::production(
self.context.service_context(format!("node_{}", index)),
beacon_config,
)
.map(move |beacon_node| {
self_1.beacon_nodes.write().push(beacon_node);
})
}
/// Adds a validator client to the network, connecting it to the beacon node with index
/// `beacon_node`.
pub fn add_validator_client(
&self,
mut validator_config: ValidatorConfig,
beacon_node: usize,
keypair_indices: Vec<usize>,
) -> impl Future<Item = (), Error = String> {
let index = self.validator_clients.read().len();
let context = self.context.service_context(format!("validator_{}", index));
let self_1 = self.clone();
self.beacon_nodes
.read()
.get(beacon_node)
.map(move |beacon_node| {
let socket_addr = beacon_node
.client
.http_listen_addr()
.expect("Must have http started");
validator_config.http_server =
format!("http://{}:{}", socket_addr.ip(), socket_addr.port());
validator_config
})
.ok_or_else(|| format!("No beacon node for index {}", beacon_node))
.into_future()
.and_then(move |validator_config| {
LocalValidatorClient::production_with_insecure_keypairs(
context,
validator_config,
&keypair_indices,
)
})
.map(move |validator_client| self_1.validator_clients.write().push(validator_client))
}
/// For all beacon nodes in `Self`, return a HTTP client to access each nodes HTTP API.
pub fn remote_nodes(&self) -> Result<Vec<RemoteBeaconNode<E>>, String> {
let beacon_nodes = self.beacon_nodes.read();
beacon_nodes
.iter()
.map(|beacon_node| beacon_node.remote_node())
.collect()
}
/// Return current epoch of bootnode.
pub fn bootnode_epoch(&self) -> impl Future<Item = Epoch, Error = String> {
let nodes = self.remote_nodes().expect("Failed to get remote nodes");
let bootnode = nodes.first().expect("Should contain bootnode");
bootnode
.http
.beacon()
.get_head()
.map_err(|e| format!("Cannot get head: {:?}", e))
.map(|head| head.finalized_slot.epoch(E::slots_per_epoch()))
}
}

380
tests/simulator/src/main.rs Normal file
View File

@@ -0,0 +1,380 @@
//! This crate provides a simluation that creates `n` beacon node and validator clients, each with
//! `v` validators. A deposit contract is deployed at the start of the simulation using a local
//! `ganache-cli` instance (you must have `ganache-cli` installed and avaliable on your path). All
//! beacon nodes independently listen for genesis from the deposit contract, then start operating.
//!
//! As the simulation runs, there are checks made to ensure that all components are running
//! correctly. If any of these checks fail, the simulation will exit immediately.
//!
//! ## Future works
//!
//! Presently all the beacon nodes and validator clients all log to stdout. Additionally, the
//! simulation uses `println` to communicate some info. It might be nice if the nodes logged to
//! easy-to-find files and stdout only contained info from the simulation.
//!
#[macro_use]
extern crate clap;
mod checks;
mod cli;
mod local_network;
mod sync_sim;
use clap::ArgMatches;
use cli::cli_app;
use env_logger::{Builder, Env};
use eth1_test_rig::GanacheEth1Instance;
use futures::{future, stream, Future, Stream};
use local_network::LocalNetwork;
use node_test_rig::{
environment::EnvironmentBuilder, testing_client_config, ClientGenesis, ValidatorConfig,
};
use std::time::{Duration, Instant};
use sync_sim::*;
use tokio::timer::Interval;
use types::MinimalEthSpec;
pub type E = MinimalEthSpec;
fn main() {
// Debugging output for libp2p and external crates.
Builder::from_env(Env::default()).init();
let matches = cli_app().get_matches();
match matches.subcommand() {
("beacon-chain-sim", Some(matches)) => match run_beacon_chain_sim(matches) {
Ok(()) => println!("Simulation exited successfully"),
Err(e) => {
eprintln!("Simulation exited with error: {}", e);
std::process::exit(1)
}
},
("syncing-sim", Some(matches)) => match run_syncing_sim(matches) {
Ok(()) => println!("Simulation exited successfully"),
Err(e) => {
eprintln!("Simulation exited with error: {}", e);
std::process::exit(1)
}
},
_ => {
eprintln!("Invalid subcommand. Use --help to see available options");
std::process::exit(1)
}
}
}
fn run_beacon_chain_sim(matches: &ArgMatches) -> Result<(), String> {
let nodes = value_t!(matches, "nodes", usize).unwrap_or(4);
let validators_per_node = value_t!(matches, "validators_per_node", usize).unwrap_or(20);
let speed_up_factor = value_t!(matches, "nodes", u64).unwrap_or(4);
let mut end_after_checks = true;
if matches.is_present("end_after_checks") {
end_after_checks = false;
}
println!("Beacon Chain Simulator:");
println!(" nodes:{}", nodes);
println!(" validators_per_node:{}", validators_per_node);
println!(" end_after_checks:{}", end_after_checks);
let log_level = "debug";
let log_format = None;
beacon_chain_sim(
nodes,
validators_per_node,
speed_up_factor,
log_level,
log_format,
end_after_checks,
)
}
fn run_syncing_sim(matches: &ArgMatches) -> Result<(), String> {
let initial_delay = value_t!(matches, "initial_delay", u64).unwrap_or(50);
let sync_delay = value_t!(matches, "sync_delay", u64).unwrap_or(10);
let speed_up_factor = value_t!(matches, "speedup", u64).unwrap_or(15);
let strategy = value_t!(matches, "strategy", String).unwrap_or("all".into());
println!("Syncing Simulator:");
println!(" initial_delay:{}", initial_delay);
println!(" sync delay:{}", sync_delay);
println!(" speed up factor:{}", speed_up_factor);
println!(" strategy:{}", strategy);
let log_level = "debug";
let log_format = None;
syncing_sim(
speed_up_factor,
initial_delay,
sync_delay,
strategy,
log_level,
log_format,
)
}
fn syncing_sim(
speed_up_factor: u64,
initial_delay: u64,
sync_delay: u64,
strategy: String,
log_level: &str,
log_format: Option<&str>,
) -> Result<(), String> {
let mut env = EnvironmentBuilder::minimal()
.async_logger(log_level, log_format)?
.multi_threaded_tokio_runtime()?
.build()?;
let spec = &mut env.eth2_config.spec;
let end_after_checks = true;
spec.milliseconds_per_slot = spec.milliseconds_per_slot / speed_up_factor;
spec.min_genesis_time = 0;
spec.min_genesis_active_validator_count = 16;
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
let context = env.core_context();
let beacon_config = testing_client_config();
let num_validators = 8;
let future = LocalNetwork::new(context, beacon_config.clone())
/*
* Add a validator client which handles all validators from the genesis state.
*/
.and_then(move |network| {
network
.add_validator_client(ValidatorConfig::default(), 0, (0..num_validators).collect())
.map(|_| network)
})
/*
* Start the processes that will run checks on the network as it runs.
*/
.and_then(move |network| {
// The `final_future` either completes immediately or never completes, depending on the value
// of `end_after_checks`.
let final_future: Box<dyn Future<Item = (), Error = String> + Send> =
if end_after_checks {
Box::new(future::ok(()).map_err(|()| "".to_string()))
} else {
Box::new(future::empty().map_err(|()| "".to_string()))
};
future::ok(())
// Check all syncing strategies one after other.
.join(pick_strategy(
&strategy,
network.clone(),
beacon_config.clone(),
slot_duration,
initial_delay,
sync_delay,
))
.join(final_future)
.map(|_| network)
})
/*
* End the simulation by dropping the network. This will kill all running beacon nodes and
* validator clients.
*/
.map(|network| {
println!(
"Simulation complete. Finished with {} beacon nodes and {} validator clients",
network.beacon_node_count(),
network.validator_client_count()
);
// Be explicit about dropping the network, as this kills all the nodes. This ensures
// all the checks have adequate time to pass.
drop(network)
});
env.runtime().block_on(future)
}
fn beacon_chain_sim(
node_count: usize,
validators_per_node: usize,
speed_up_factor: u64,
log_level: &str,
log_format: Option<&str>,
end_after_checks: bool,
) -> Result<(), String> {
let mut env = EnvironmentBuilder::minimal()
.async_logger(log_level, log_format)?
.multi_threaded_tokio_runtime()?
.build()?;
let eth1_block_time = Duration::from_millis(15_000 / speed_up_factor);
let spec = &mut env.eth2_config.spec;
spec.milliseconds_per_slot /= speed_up_factor;
spec.eth1_follow_distance = 16;
spec.seconds_per_day = eth1_block_time.as_secs() * spec.eth1_follow_distance * 2;
spec.min_genesis_time = 0;
spec.min_genesis_active_validator_count = 64;
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
let initial_validator_count = spec.min_genesis_active_validator_count as usize;
let total_validator_count = validators_per_node * node_count;
let deposit_amount = env.eth2_config.spec.max_effective_balance;
let context = env.core_context();
let executor = context.executor.clone();
let future = GanacheEth1Instance::new()
/*
* Deploy the deposit contract, spawn tasks to keep creating new blocks and deposit
* validators.
*/
.map(move |ganache_eth1_instance| {
let deposit_contract = ganache_eth1_instance.deposit_contract;
let ganache = ganache_eth1_instance.ganache;
let eth1_endpoint = ganache.endpoint();
let deposit_contract_address = deposit_contract.address();
// Start a timer that produces eth1 blocks on an interval.
executor.spawn(
Interval::new(Instant::now(), eth1_block_time)
.map_err(|_| eprintln!("Eth1 block timer failed"))
.for_each(move |_| ganache.evm_mine().map_err(|_| ()))
.map_err(|_| eprintln!("Eth1 evm_mine failed"))
.map(|_| ()),
);
// Submit deposits to the deposit contract.
executor.spawn(
stream::unfold(0..total_validator_count, move |mut iter| {
iter.next().map(|i| {
println!("Submitting deposit for validator {}...", i);
deposit_contract
.deposit_deterministic_async::<E>(i, deposit_amount)
.map(|_| ((), iter))
})
})
.collect()
.map(|_| ())
.map_err(|e| eprintln!("Error submitting deposit: {}", e)),
);
let mut beacon_config = testing_client_config();
beacon_config.genesis = ClientGenesis::DepositContract;
beacon_config.eth1.endpoint = eth1_endpoint;
beacon_config.eth1.deposit_contract_address = deposit_contract_address;
beacon_config.eth1.deposit_contract_deploy_block = 0;
beacon_config.eth1.lowest_cached_block_number = 0;
beacon_config.eth1.follow_distance = 1;
beacon_config.dummy_eth1_backend = false;
beacon_config.sync_eth1_chain = true;
beacon_config
})
/*
* Create a new `LocalNetwork` with one beacon node.
*/
.and_then(move |beacon_config| {
LocalNetwork::new(context, beacon_config.clone())
.map(|network| (network, beacon_config))
})
/*
* One by one, add beacon nodes to the network.
*/
.and_then(move |(network, beacon_config)| {
let network_1 = network.clone();
stream::unfold(0..node_count - 1, move |mut iter| {
iter.next().map(|_| {
network_1
.add_beacon_node(beacon_config.clone())
.map(|()| ((), iter))
})
})
.collect()
.map(|_| network)
})
/*
* One by one, add validator clients to the network. Each validator client is attached to
* a single corresponding beacon node.
*/
.and_then(move |network| {
let network_1 = network.clone();
// Note: presently the validator client future will only resolve once genesis time
// occurs. This is great for this scenario, but likely to change in the future.
//
// If the validator client future behaviour changes, we would need to add a new future
// that delays until genesis. Otherwise, all of the checks that start in the next
// future will start too early.
stream::unfold(0..node_count, move |mut iter| {
iter.next().map(|i| {
let indices = (i * validators_per_node..(i + 1) * validators_per_node)
.collect::<Vec<_>>();
network_1
.add_validator_client(ValidatorConfig::default(), i, indices)
.map(|()| ((), iter))
})
})
.collect()
.map(|_| network)
})
/*
* Start the processes that will run checks on the network as it runs.
*/
.and_then(move |network| {
// The `final_future` either completes immediately or never completes, depending on the value
// of `end_after_checks`.
let final_future: Box<dyn Future<Item = (), Error = String> + Send> =
if end_after_checks {
Box::new(future::ok(()).map_err(|()| "".to_string()))
} else {
Box::new(future::empty().map_err(|()| "".to_string()))
};
future::ok(())
// Check that the chain finalizes at the first given opportunity.
.join(checks::verify_first_finalization(
network.clone(),
slot_duration,
))
// Check that the chain starts with the expected validator count.
.join(checks::verify_initial_validator_count(
network.clone(),
slot_duration,
initial_validator_count,
))
// Check that validators greater than `spec.min_genesis_active_validator_count` are
// onboarded at the first possible opportunity.
.join(checks::verify_validator_onboarding(
network.clone(),
slot_duration,
total_validator_count,
))
// End now or run forever, depending on the `end_after_checks` flag.
.join(final_future)
.map(|_| network)
})
/*
* End the simulation by dropping the network. This will kill all running beacon nodes and
* validator clients.
*/
.map(|network| {
println!(
"Simulation complete. Finished with {} beacon nodes and {} validator clients",
network.beacon_node_count(),
network.validator_client_count()
);
// Be explicit about dropping the network, as this kills all the nodes. This ensures
// all the checks have adequate time to pass.
drop(network)
});
env.runtime().block_on(future)
}

201
tests/simulator/src/sync_sim.rs Normal file
View File

@@ -0,0 +1,201 @@
use crate::checks::{epoch_delay, verify_all_finalized_at};
use crate::local_network::LocalNetwork;
use futures::{Future, IntoFuture};
use node_test_rig::ClientConfig;
use std::time::Duration;
use types::{Epoch, EthSpec};
pub fn pick_strategy<E: EthSpec>(
strategy: &str,
network: LocalNetwork<E>,
beacon_config: ClientConfig,
slot_duration: Duration,
initial_delay: u64,
sync_delay: u64,
) -> Box<dyn Future<Item = (), Error = String> + Send + 'static> {
match strategy {
"one-node" => Box::new(verify_one_node_sync(
network,
beacon_config,
slot_duration,
initial_delay,
sync_delay,
)),
"two-nodes" => Box::new(verify_two_nodes_sync(
network,
beacon_config,
slot_duration,
initial_delay,
sync_delay,
)),
"mixed" => Box::new(verify_in_between_sync(
network,
beacon_config,
slot_duration,
initial_delay,
sync_delay,
)),
"all" => Box::new(verify_syncing(
network,
beacon_config,
slot_duration,
initial_delay,
sync_delay,
)),
_ => Box::new(Err("Invalid strategy".into()).into_future()),
}
}
/// Verify one node added after `initial_delay` epochs is in sync
/// after `sync_delay` epochs.
pub fn verify_one_node_sync<E: EthSpec>(
network: LocalNetwork<E>,
beacon_config: ClientConfig,
slot_duration: Duration,
initial_delay: u64,
sync_delay: u64,
) -> impl Future<Item = (), Error = String> {
// Delay for `initial_delay` epochs before adding another node to start syncing
epoch_delay(
Epoch::new(initial_delay),
slot_duration,
E::slots_per_epoch(),
)
.and_then(move |_| {
// Add a beacon node
network.add_beacon_node(beacon_config).map(|_| network)
})
.and_then(move |network| {
// Delay for `sync_delay` epochs before verifying synced state.
epoch_delay(Epoch::new(sync_delay), slot_duration, E::slots_per_epoch()).map(|_| network)
})
.and_then(move |network| network.bootnode_epoch().map(|e| (e, network)))
.and_then(move |(epoch, network)| {
verify_all_finalized_at(network, epoch).map_err(|e| format!("One node sync error: {}", e))
})
}
/// Verify two nodes added after `initial_delay` epochs are in sync
/// after `sync_delay` epochs.
pub fn verify_two_nodes_sync<E: EthSpec>(
network: LocalNetwork<E>,
beacon_config: ClientConfig,
slot_duration: Duration,
initial_delay: u64,
sync_delay: u64,
) -> impl Future<Item = (), Error = String> {
// Delay for `initial_delay` epochs before adding another node to start syncing
epoch_delay(
Epoch::new(initial_delay),
slot_duration,
E::slots_per_epoch(),
)
.and_then(move |_| {
// Add beacon nodes
network
.add_beacon_node(beacon_config.clone())
.join(network.add_beacon_node(beacon_config.clone()))
.map(|_| network)
})
.and_then(move |network| {
// Delay for `sync_delay` epochs before verifying synced state.
epoch_delay(Epoch::new(sync_delay), slot_duration, E::slots_per_epoch()).map(|_| network)
})
.and_then(move |network| network.bootnode_epoch().map(|e| (e, network)))
.and_then(move |(epoch, network)| {
verify_all_finalized_at(network, epoch).map_err(|e| format!("Two node sync error: {}", e))
})
}
/// Add 2 syncing nodes after `initial_delay` epochs,
/// Add another node after `sync_delay - 5` epochs and verify all are
/// in sync after `sync_delay + 5` epochs.
pub fn verify_in_between_sync<E: EthSpec>(
network: LocalNetwork<E>,
beacon_config: ClientConfig,
slot_duration: Duration,
initial_delay: u64,
sync_delay: u64,
) -> impl Future<Item = (), Error = String> {
// Delay for `initial_delay` epochs before adding another node to start syncing
let config1 = beacon_config.clone();
epoch_delay(
Epoch::new(initial_delay),
slot_duration,
E::slots_per_epoch(),
)
.and_then(move |_| {
// Add a beacon node
network
.add_beacon_node(beacon_config.clone())
.join(network.add_beacon_node(beacon_config.clone()))
.map(|_| network)
})
.and_then(move |network| {
// Delay before adding additional syncing nodes.
epoch_delay(
Epoch::new(sync_delay - 5),
slot_duration,
E::slots_per_epoch(),
)
.map(|_| network)
})
.and_then(move |network| {
// Add a beacon node
network.add_beacon_node(config1.clone()).map(|_| network)
})
.and_then(move |network| {
// Delay for `sync_delay` epochs before verifying synced state.
epoch_delay(
Epoch::new(sync_delay + 5),
slot_duration,
E::slots_per_epoch(),
)
.map(|_| network)
})
.and_then(move |network| network.bootnode_epoch().map(|e| (e, network)))
.and_then(move |(epoch, network)| {
verify_all_finalized_at(network, epoch).map_err(|e| format!("In between sync error: {}", e))
})
}
/// Run syncing strategies one after other.
pub fn verify_syncing<E: EthSpec>(
network: LocalNetwork<E>,
beacon_config: ClientConfig,
slot_duration: Duration,
initial_delay: u64,
sync_delay: u64,
) -> impl Future<Item = (), Error = String> {
verify_one_node_sync(
network.clone(),
beacon_config.clone(),
slot_duration,
initial_delay,
sync_delay,
)
.map(|_| println!("Completed one node sync"))
.and_then(move |_| {
verify_two_nodes_sync(
network.clone(),
beacon_config.clone(),
slot_duration,
initial_delay,
sync_delay,
)
.map(|_| {
println!("Completed two node sync");
(network, beacon_config)
})
})
.and_then(move |(network, beacon_config)| {
verify_in_between_sync(
network,
beacon_config,
slot_duration,
initial_delay,
sync_delay,
)
.map(|_| println!("Completed in between sync"))
})
}