use anyhow::anyhow;
use clap::{
Args,
builder::ArgPredicate::IsPresent,
};
use fuel_core::{
p2p::{
Multiaddr,
config::{
Config,
MAX_RESPONSE_SIZE,
NotInitialized,
convert_to_libp2p_keypair,
},
gossipsub_config::default_gossipsub_builder,
heartbeat,
},
types::{
fuel_crypto,
fuel_crypto::SecretKey,
},
};
use std::{
net::{
IpAddr,
Ipv4Addr,
},
num::{
NonZeroU32,
NonZeroUsize,
},
path::PathBuf,
str::FromStr,
};
const MAX_RESPONSE_SIZE_STR: &str =
const_format::formatcp!("{}", MAX_RESPONSE_SIZE.get());
#[derive(Debug, Clone, Args)]
pub struct P2PArgs {
#[clap(long = "enable-p2p", action)]
pub enable_p2p: bool,
#[clap(long = "keypair", env, value_parser = KeypairArg::try_from_string)]
#[arg(required_if_eq("enable_p2p", "true"))]
#[arg(requires_if(IsPresent, "enable_p2p"))]
pub keypair: Option<KeypairArg>,
#[clap(long = "address", env)]
pub address: Option<IpAddr>,
#[clap(long = "public-address", env)]
pub public_address: Option<Multiaddr>,
#[clap(long = "peering-port", default_value = "30333", env)]
pub peering_port: u16,
#[clap(long = "max-block-size", default_value = MAX_RESPONSE_SIZE_STR, env)]
pub max_block_size: u32,
#[clap(long = "max-headers-per-request", default_value = "100", env)]
pub max_headers_per_request: usize,
#[clap(long = "max-txs-per-request", default_value = "10000", env)]
pub max_txs_per_request: usize,
#[clap(long = "bootstrap-nodes", value_delimiter = ',', env)]
pub bootstrap_nodes: Vec<Multiaddr>,
#[clap(long = "reserved-nodes", value_delimiter = ',', env)]
pub reserved_nodes: Vec<Multiaddr>,
#[clap(long = "reserved-nodes-only-mode", env)]
pub reserved_nodes_only_mode: bool,
#[clap(long = "enable-mdns", env)]
pub enable_mdns: bool,
#[clap(long = "max-peers-connected", default_value = "50", env)]
pub max_peers_connected: u32,
#[clap(long = "max-discovery-peers-connected", default_value = "10000", env)]
pub max_discovery_peers_connected: u32,
#[clap(long = "max-outgoing-connections", default_value = "10", env)]
pub max_outgoing_connections: u32,
#[clap(long = "max-connections-per-peer", default_value = "3", env)]
pub max_connections_per_peer: Option<u32>,
#[clap(long = "random-walk", env)]
pub random_walk: Option<humantime::Duration>,
#[clap(long = "allow-private-addresses", env)]
pub allow_private_addresses: bool,
#[clap(long = "connection-idle-timeout", default_value = "120s", env)]
pub connection_idle_timeout: humantime::Duration,
#[clap(long = "info-interval", default_value = "3s", env)]
pub info_interval: humantime::Duration,
#[clap(long = "identify-interval", default_value = "5s", env)]
pub identify_interval: humantime::Duration,
#[clap(long = "max-mesh-size", default_value = "12", env)]
pub max_mesh_size: usize,
#[clap(long = "min-mesh-size", default_value = "4", env)]
pub min_mesh_size: usize,
#[clap(long = "ideal-mesh-size", default_value = "6", env)]
pub ideal_mesh_size: usize,
#[clap(long = "history-length", default_value = "5", env)]
pub history_length: usize,
#[clap(long = "history-gossip", default_value = "3", env)]
pub history_gossip: usize,
#[clap(long = "gossip-heartbeat-interval", default_value = "500ms", env)]
pub gossip_heartbeat_interval: humantime::Duration,
#[clap(long = "max-transmit-size", default_value = MAX_RESPONSE_SIZE_STR, env)]
pub max_transmit_size: usize,
#[clap(long = "request-timeout", default_value = "20s", env)]
pub request_timeout: humantime::Duration,
#[clap(long = "request-max-concurrent-streams", default_value = "256", env)]
pub max_concurrent_streams: usize,
#[clap(long = "connection-keep-alive", default_value = "20s", env)]
pub connection_keep_alive: humantime::Duration,
#[clap(long = "heartbeat-send-duration", default_value = "2s", env)]
pub heartbeat_send_duration: humantime::Duration,
#[clap(long = "heartbeat-idle-duration", default_value = "100ms", env)]
pub heartbeat_idle_duration: humantime::Duration,
#[clap(long = "heartbeat-max-failures", default_value = "5", env)]
pub heartbeat_max_failures: NonZeroU32,
#[clap(long = "heartbeat-check-interval", default_value = "5s", env)]
pub heartbeat_check_interval: humantime::Duration,
#[clap(long = "heartbeat-max-avg-interval", default_value = "20s", env)]
pub heartbeat_max_avg_interval: humantime::Duration,
#[clap(long = "heartbeat-max-time-since-last", default_value = "40s", env)]
pub heartbeat_max_time_since_last: humantime::Duration,
#[clap(long = "p2p-database-read-threads", default_value = "2", env)]
pub database_read_threads: usize,
#[clap(long = "p2p-txpool-threads", default_value = "0", env)]
pub tx_pool_threads: usize,
#[clap(long = "subscribe-to-pre-confirmations", env)]
subscribe_to_pre_confirmations: bool,
#[clap(long = "subscribe-to-transactions", env)]
subscribe_to_transactions: bool,
#[clap(long = "p2p-cache-size", default_value = "1000", env)]
pub cache_size: Option<NonZeroUsize>,
}
#[derive(Debug, Clone, Args)]
pub struct SyncArgs {
#[clap(long = "sync-block-stream-buffer-size", default_value = "10", env)]
pub block_stream_buffer_size: usize,
#[clap(long = "sync-header-batch-size", default_value = "10", env)]
pub header_batch_size: u32,
}
#[derive(Clone, Debug)]
pub enum KeypairArg {
Path(PathBuf),
InlineSecret(SecretKey),
}
impl KeypairArg {
pub fn try_from_string(s: &str) -> anyhow::Result<KeypairArg> {
let secret = SecretKey::from_str(s);
if let Ok(secret) = secret {
return Ok(KeypairArg::InlineSecret(secret))
}
let path = PathBuf::from_str(s);
let Ok(pathbuf) = path;
if pathbuf.exists() {
Ok(KeypairArg::Path(pathbuf))
} else {
Err(anyhow!(
"path `{pathbuf:?}` does not exist for keypair argument"
))
}
}
}
impl From<SyncArgs> for fuel_core::sync::Config {
fn from(value: SyncArgs) -> Self {
Self {
block_stream_buffer_size: value.block_stream_buffer_size,
header_batch_size: value.header_batch_size as usize,
}
}
}
impl P2PArgs {
pub fn into_config(
self,
network_name: String,
metrics: bool,
) -> anyhow::Result<Option<Config<NotInitialized>>> {
if !self.enable_p2p {
tracing::info!("P2P service disabled");
return Ok(None)
}
let local_keypair = {
match self.keypair.expect("mandatory value") {
KeypairArg::Path(path) => {
let phrase = std::fs::read_to_string(path)?;
let secret_key =
fuel_crypto::SecretKey::new_from_mnemonic_phrase_with_path(
&phrase,
"m/44'/60'/0'/0/0",
)?;
convert_to_libp2p_keypair(&mut secret_key.to_vec())?
}
KeypairArg::InlineSecret(secret_key) => {
convert_to_libp2p_keypair(&mut secret_key.to_vec())?
}
}
};
let gossipsub_config = default_gossipsub_builder()
.mesh_n(self.ideal_mesh_size)
.mesh_n_low(self.min_mesh_size)
.mesh_n_high(self.max_mesh_size)
.history_length(self.history_length)
.history_gossip(self.history_gossip)
.heartbeat_interval(self.gossip_heartbeat_interval.into())
.max_transmit_size(self.max_transmit_size)
.build()
.expect("valid gossipsub configuration");
let random_walk = self.random_walk.map(Into::into);
let heartbeat_config = {
let send_duration = self.heartbeat_send_duration.into();
let idle_duration = self.heartbeat_idle_duration.into();
heartbeat::Config::new(
send_duration,
idle_duration,
self.heartbeat_max_failures,
)
};
let max_block_size = NonZeroU32::new(self.max_block_size)
.ok_or(anyhow!("max block size must be greater than zero"))?;
let config = Config {
keypair: local_keypair,
network_name,
checksum: Default::default(),
address: self
.address
.unwrap_or_else(|| IpAddr::V4(Ipv4Addr::from([0, 0, 0, 0]))),
public_address: self.public_address,
tcp_port: self.peering_port,
max_block_size,
max_headers_per_request: self.max_headers_per_request,
max_txs_per_request: self.max_txs_per_request,
bootstrap_nodes: self.bootstrap_nodes,
reserved_nodes: self.reserved_nodes,
reserved_nodes_only_mode: self.reserved_nodes_only_mode,
enable_mdns: self.enable_mdns,
max_discovery_peers_connected: self.max_discovery_peers_connected,
max_outgoing_connections: self.max_outgoing_connections,
max_connections_per_peer: self.max_connections_per_peer,
max_functional_peers_connected: self.max_peers_connected,
allow_private_addresses: self.allow_private_addresses,
random_walk,
connection_idle_timeout: Some(self.connection_idle_timeout.into()),
gossipsub_config,
heartbeat_config,
set_request_timeout: self.request_timeout.into(),
max_concurrent_streams: self.max_concurrent_streams,
set_connection_keep_alive: self.connection_keep_alive.into(),
heartbeat_check_interval: self.heartbeat_check_interval.into(),
heartbeat_max_avg_interval: self.heartbeat_max_avg_interval.into(),
heartbeat_max_time_since_last: self.heartbeat_max_time_since_last.into(),
info_interval: Some(self.info_interval.into()),
identify_interval: Some(self.identify_interval.into()),
metrics,
database_read_threads: self.database_read_threads,
tx_pool_threads: self.tx_pool_threads,
state: NotInitialized,
subscribe_to_pre_confirmations: self.subscribe_to_pre_confirmations,
subscribe_to_transactions: self.subscribe_to_transactions,
cache_size: self.cache_size,
};
Ok(Some(config))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_invalid_path() {
let invalid_path = "/invalid/path/to/keypair";
let keypair = KeypairArg::try_from_string(invalid_path);
let err = keypair.expect_err("The path is incorrect it should fail");
assert!(
err.to_string()
.contains("does not exist for keypair argument")
);
}
}