Merge pull request #38 from rustaceanrob/8-18-conf

Add `configure_me` dependency for configurations

Author: rustaceanrob

node/Cargo.toml

@@ -2,6 +2,7 @@
 name = "node"
 version = "0.1.0"
 edition = "2021"
+build = "build.rs"
 
 [dependencies]
 accumulator = { path = "../accumulator/" }
@@ -11,8 +12,15 @@ hintfile = { path = "../hintfile/" }
 network = { path = "../network/" }
 p2p = { workspace = true }
 
+configure_me = "0.4.0"
 tracing = "0.1"
 tracing-subscriber = "0.3"
 
+[build-dependencies]
+configure_me_codegen = "0.4.0"
+
+[package.metadata.configure_me]
+spec = "config_spec.toml"
+
 [[bin]]
 name = "ibd"

node/build.rs

@@ -0,0 +1,5 @@
+extern crate configure_me_codegen;
+
+fn main() -> Result<(), configure_me_codegen::Error> {
+    configure_me_codegen::build_script_auto()
+}

node/config_spec.toml

@@ -0,0 +1,41 @@
+[[param]]
+name = "hintfile"
+type = "String"
+default = "\"./bitcoin.hints\".into()"
+doc = "The path to your `bitcoin.hints` file that will be used for IBD"
+
+[[param]]
+name = "blocks_dir"
+type = "String"
+default = "\"./blockfiles\".into()"
+doc = "Directory where you would like to store the bitcoin blocks"
+
+[[param]]
+name = "network"
+type = "String"
+default = "\"bitcoin\".into()"
+doc = "The bitcoin network to operate on. Options are `bitcoin` or `signet`"
+
+[[param]]
+name = "ping_timeout"
+type = "u64"
+default = "15"
+doc = "The time a peer has to respond to a `ping` message. Pings are sent aggressively throughout IBD to find slow peers."
+
+[[param]]
+name = "tcp_timeout"
+type = "u64"
+default = "2"
+doc = "The maximum time to establish a connection"
+
+[[param]]
+name = "read_timeout"
+type = "u64"
+default = "2"
+doc = "The maximum time to read from a TCP stream until the connection is killed."
+
+[[param]]
+name = "write_timeout"
+type = "u64"
+default = "2"
+doc = "The maximum time to write to a TCP stream until the connection is killed."

node/src/bin/ibd.rs

@@ -1,49 +1,62 @@
 use std::{
     fs::File,
-    path::Path,
+    path::PathBuf,
     sync::{mpsc::channel, Arc, Mutex},
-    time::Instant,
+    time::{Duration, Instant},
 };
 
 use bitcoin::{consensus, BlockHash, Network};
 use hintfile::Hints;
-use kernel::{ChainType, ChainstateManager, ChainstateManagerOptions, ContextBuilder};
+use kernel::{ChainstateManager, ChainstateManagerOptions, ContextBuilder};
 
 use node::{
     bootstrap_dns, elapsed_time, get_blocks_for_range, hashes_from_chain, sync_block_headers,
-    AccumulatorState,
+    AccumulatorState, ChainExt,
 };
+use p2p::net::TimeoutParams;
 
-const CHAIN_TYPE: ChainType = ChainType::SIGNET;
-const NETWORK: Network = Network::Signet;
 const TASKS: usize = 256;
-const BLOCK_FILE_PATH: &str = "./blockfiles";
+const PING_INTERVAL: Duration = Duration::from_secs(15);
+
+configure_me::include_config!();
 
 fn main() {
-    let mut args = std::env::args();
-    let _ = args.next();
-    let hint_path = args.next().expect("Usage: <path_to_hints_file>");
-    // Logging
+    let (config, _) = Config::including_optional_config_files::<&[&str]>(&[]).unwrap_or_exit();
+    let hint_path = config.hintfile;
+    let blocks_dir = config.blocks_dir;
+    let network = config
+        .network
+        .parse::<Network>()
+        .expect("invalid network string");
+    let ping_timeout = Duration::from_secs(config.ping_timeout);
+    let tcp_timeout = Duration::from_secs(config.tcp_timeout);
+    let read_timeout = Duration::from_secs(config.read_timeout);
+    let write_timeout = Duration::from_secs(config.write_timeout);
+    let mut timeout_conf = TimeoutParams::new();
+    timeout_conf.read_timeout(read_timeout);
+    timeout_conf.write_timeout(write_timeout);
+    timeout_conf.tcp_handshake_timeout(tcp_timeout);
+    timeout_conf.ping_interval(PING_INTERVAL);
     let subscriber = tracing_subscriber::FmtSubscriber::new();
     tracing::subscriber::set_global_default(subscriber).unwrap();
     let hintfile_start_time = Instant::now();
     tracing::info!("Reading in {hint_path}");
     let mut hintfile = File::open(hint_path).expect("invalid hintfile path");
     let hints = Arc::new(Hints::from_file(&mut hintfile));
     elapsed_time(hintfile_start_time);
-    let block_file_path = Path::new(BLOCK_FILE_PATH);
-    std::fs::create_dir(block_file_path).expect("could not create block file directory");
+    let block_file_path = PathBuf::from(&blocks_dir);
+    std::fs::create_dir(&block_file_path).expect("could not create block file directory");
     let stop_hash =
         consensus::deserialize::<BlockHash>(&hints.stop_hash()).expect("stop hash is not valid");
     tracing::info!("Assume valid hash: {stop_hash}");
     tracing::info!("Finding peers with DNS");
     let dns_start_time = Instant::now();
-    let peers = bootstrap_dns(NETWORK);
+    let peers = bootstrap_dns(network);
     elapsed_time(dns_start_time);
     tracing::info!("Initializing bitcoin kernel");
     let kernel_start_time = Instant::now();
     let ctx = ContextBuilder::new()
-        .chain_type(CHAIN_TYPE)
+        .chain_type(network.chain_type())
         .build()
         .unwrap();
     let options = ChainstateManagerOptions::new(&ctx, ".", "./blocks").unwrap();
@@ -52,7 +65,7 @@ fn main() {
     let tip = chainman.best_header().height();
     tracing::info!("Kernel best header: {tip}");
     let chain = Arc::new(chainman);
-    sync_block_headers(stop_hash, &peers, Arc::clone(&chain), NETWORK);
+    sync_block_headers(stop_hash, &peers, Arc::clone(&chain), network, timeout_conf);
     tracing::info!("Assume valid height: {}", chain.best_header().height());
     let (tx, rx) = channel();
     let main_routine_time = Instant::now();
@@ -67,11 +80,14 @@ fn main() {
         let tx = tx.clone();
         let peers = Arc::clone(&peers);
         let hints = Arc::clone(&hints);
+        let block_file_path = block_file_path.clone();
         let block_task = std::thread::spawn(move || {
             get_blocks_for_range(
                 task_id as u32,
-                NETWORK,
-                block_file_path,
+                timeout_conf,
+                ping_timeout,
+                network,
+                &block_file_path,
                 chain,
                 &hints,
                 peers,

node/src/lib.rs

@@ -20,7 +20,7 @@ use bitcoin::{
     BlockHash, Network, OutPoint,
 };
 use hintfile::Hints;
-use kernel::ChainstateManager;
+use kernel::{ChainType, ChainstateManager};
 use network::dns::DnsQuery;
 use p2p::{
     handshake::ConnectionConfig,
@@ -79,6 +79,7 @@ pub fn sync_block_headers(
     hosts: &[SocketAddr],
     chainman: Arc<ChainstateManager>,
     network: Network,
+    timeout_params: TimeoutParams,
 ) {
     let mut rng = thread_rng();
     let then = Instant::now();
@@ -89,13 +90,10 @@ pub fn sync_block_headers(
             .copied()
             .expect("dns must return at least one peer");
         tracing::info!("Attempting connection to {random}");
-        let mut timeout_conf = TimeoutParams::new();
-        timeout_conf.read_timeout(Duration::from_secs(2));
-        timeout_conf.write_timeout(Duration::from_secs(2));
         let conn = ConnectionConfig::new()
             .change_network(network)
             .decrease_version_requirement(ProtocolVersion::BIP0031_VERSION)
-            .open_connection(random, timeout_conf);
+            .open_connection(random, timeout_params);
         let (writer, mut reader, metrics) = match conn {
             Ok((writer, reader, metrics)) => (writer, reader, metrics),
             Err(_) => continue,
@@ -153,6 +151,8 @@ pub fn sync_block_headers(
 #[allow(clippy::too_many_arguments)]
 pub fn get_blocks_for_range(
     task_id: u32,
+    timeout_params: TimeoutParams,
+    ping_timeout: Duration,
     network: Network,
     block_dir: &Path,
     chain: Arc<ChainstateManager>,
@@ -161,10 +161,6 @@ pub fn get_blocks_for_range(
     updater: Sender<AccumulatorUpdate>,
     mut batch: Vec<BlockHash>,
 ) {
-    let mut timeout = TimeoutParams::new();
-    timeout.read_timeout(Duration::from_secs(2));
-    timeout.tcp_handshake_timeout(Duration::from_secs(2));
-    timeout.ping_interval(Duration::from_secs(15));
     let mut rng = thread_rng();
     loop {
         let peer = {
@@ -179,7 +175,7 @@ pub fn get_blocks_for_range(
             .request_addr()
             .set_service_requirement(ServiceFlags::NETWORK)
             .decrease_version_requirement(ProtocolVersion::BIP0031_VERSION)
-            .open_connection(peer, timeout);
+            .open_connection(peer, timeout_params);
         let Ok((writer, mut reader, metrics)) = conn else {
             // tracing::warn!("Connection failed");
             continue;
@@ -269,7 +265,7 @@ pub fn get_blocks_for_range(
                 _ => (),
             }
             if let Some(message_rate) = metrics.message_rate(TimedMessage::Block) {
-                if message_rate.total_count() < 2 {
+                if message_rate.total_count() < 100 {
                     continue;
                 }
                 let Some(rate) = message_rate.messages_per_secs(Instant::now()) else {
@@ -280,6 +276,10 @@ pub fn get_blocks_for_range(
                     break;
                 }
             }
+            if metrics.ping_timed_out(ping_timeout) {
+                tracing::info!("{task_id} failed to respond to a ping");
+                break;
+            }
         }
         if batch.is_empty() {
             break;
@@ -304,3 +304,17 @@ pub fn hashes_from_chain(chain: Arc<ChainstateManager>, chunks: usize) -> Vec<Ve
     }
     hashes.chunks(chunks).map(|slice| slice.to_vec()).collect()
 }
+
+pub trait ChainExt {
+    fn chain_type(&self) -> ChainType;
+}
+
+impl ChainExt for Network {
+    fn chain_type(&self) -> ChainType {
+        match self {
+            Network::Bitcoin => ChainType::MAINNET,
+            Network::Signet => ChainType::SIGNET,
+            _ => unimplemented!("choose bitcoin or signet"),
+        }
+    }
+}