Add anchor network related cli args and wire up with network crate.

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 # Extra tooling projects will be added.
-members = ["anchor", "anchor/client", "anchor/http_api", "anchor/qbft", "anchor/p2p_network"]
+members = ["anchor", "anchor/client", "anchor/http_api", "anchor/qbft", "anchor/network", "anchor/common/version"]
 resolver = "2"
 
 [workspace.package]
@@ -10,19 +10,23 @@ edition = "2021"
 client = { path = "anchor/client" }
 qbft = { path = "anchor/qbft" }
 http_api = { path = "anchor/http_api" }
-task_executor = { git = "https://github.com/sigp/lighthouse", branch = "anchor", default-features = false, features = [
+network = { path ="anchor/network"}
+version = { path ="anchor/common/version"}
+lighthouse_network = { git = "https://github.com/sigp/lighthouse", branch = "unstable"}
+task_executor = { git = "https://github.com/sigp/lighthouse", branch = "unstable", default-features = false, features = [
     "tracing",
 ] }
-sensitive_url = { git = "https://github.com/sigp/lighthouse", branch = "anchor" }
-slot_clock = { git = "https://github.com/sigp/lighthouse", branch = "anchor" }
+sensitive_url = { git = "https://github.com/sigp/lighthouse", branch = "unstable" }
+slot_clock = { git = "https://github.com/sigp/lighthouse", branch = "unstable" }
+unused_port = { git = "https://github.com/sigp/lighthouse", branch = "unstable" }
 derive_more = { version = "1.0.0", features = ["full"] }
 async-channel = "1.9"
 axum = "0.7.7"
 clap = { version = "4.5.15", features = ["derive", "wrap_help"]}
+discv5 = "0.8.0"
 dirs = "5.0.1"
 either = "1.13.0"
 futures = "0.3.30"
-# dirs = "3"
 hyper = "1.4"
 serde = { version = "1.0.208", features = ["derive"] }
 strum = { version = "0.24", features = ["derive"] }

anchor/client/Cargo.toml

@@ -11,19 +11,16 @@ path = "src/lib.rs"
 [dependencies]
 task_executor = { workspace = true }
 http_api = { workspace = true }
+version = { workspace = true }
 clap = { workspace = true }
 serde = { workspace = true }
 strum = { workspace = true }
 sensitive_url = { workspace = true }
 dirs = { workspace = true }
 hyper = { workspace = true }
 tracing = { workspace = true }
+network = { workspace = true }
+unused_port = { workspace = true }
 # Local dependencies
 fdlimit = "0.3"
 ethereum_hashing = "0.7.0"
-git-version = "0.3.9"
-target_info = "0.1.0"
-
-[dev-dependencies]
-regex = "1.11"
-

anchor/client/src/cli.rs

@@ -4,11 +4,11 @@ use clap::{Parser, ValueEnum};
 use serde::{Deserialize, Serialize};
 use strum::Display;
 // use clap_utils::{get_color_style, FLAG_HEADER};
-use crate::version::VERSION;
 use ethereum_hashing::have_sha_extensions;
 use std::net::IpAddr;
 use std::path::PathBuf;
 use std::sync::LazyLock;
+use version::VERSION;
 
 pub static SHORT_VERSION: LazyLock<String> = LazyLock::new(|| VERSION.replace("Anchor/", ""));
 pub static LONG_VERSION: LazyLock<String> = LazyLock::new(|| {
@@ -210,6 +210,40 @@ pub struct Anchor {
     )]
     pub http_allow_origin: Option<String>,
 
+    /* Network related arguments */
+    #[clap(
+        long,
+        value_name = "ADDRESS",
+        help = "The address anchor will listen for UDP and TCP connections. To listen \
+                      over IpV4 and IpV6 set this flag twice with the different values.\n\
+                      Examples:\n\
+                      - --listen-address '0.0.0.0' will listen over IPv4.\n\
+                      - --listen-address '::' will listen over IPv6.\n\
+                      - --listen-address '0.0.0.0' --listen-address '::' will listen over both \
+                      IPv4 and IPv6. The order of the given addresses is not relevant. However, \
+                      multiple IPv4, or multiple IPv6 addresses will not be accepted.",
+        num_args(0..2),
+        default_value = "0.0.0.0",
+    )]
+    pub listen_addresses: Vec<IpAddr>,
+
+    // TODO: finish CLI
+    #[clap(skip)]
+    pub port: u16,
+    #[clap(skip)]
+    pub port6: Option<u16>,
+    #[clap(skip)]
+    pub discovery_port: Option<u16>,
+    #[clap(skip)]
+    pub discovery_port6: Option<u16>,
+    #[clap(skip)]
+    pub quic_port: Option<u16>,
+    #[clap(skip)]
+    pub quic_port6: Option<u16>,
+    // TODO: make it hidden
+    #[clap(skip)]
+    pub use_zero_ports: bool,
+
     /* Prometheus metrics HTTP server related arguments */
     #[clap(
         long,
@@ -237,7 +271,7 @@ pub struct Anchor {
         default_value_if("metrics", ArgPredicate::IsPresent, "5064"),
         requires = "metrics"
     )]
-    pub metrics_port: u16,
+    pub metrics_port: Option<u16>,
 
     #[clap(
         long,

anchor/client/src/config.rs

@@ -1,12 +1,14 @@
 // use crate::{http_api, http_metrics};
 // use clap_utils::{flags::DISABLE_MALLOC_TUNING_FLAG, parse_optional, parse_required};
 
+use crate::cli::Anchor;
+use network::{ListenAddr, ListenAddress};
 use sensitive_url::SensitiveUrl;
 use serde::{Deserialize, Serialize};
 use std::fs;
+use std::net::IpAddr;
 use std::path::PathBuf;
-
-use crate::cli::Anchor;
+use tracing::warn;
 
 pub const DEFAULT_BEACON_NODE: &str = "http://localhost:5052/";
 pub const DEFAULT_EXECUTION_NODE: &str = "http://localhost:8545/";
@@ -34,6 +36,8 @@ pub struct Config {
     pub allow_unsynced_beacon_node: bool,
     /// Configuration for the HTTP REST API.
     pub http_api: http_api::Config,
+    /// Configuration for the network stack.
+    pub network: network::Config,
     /// Configuration for the HTTP REST API.
     // TODO:
     // pub http_metrics: http_metrics::Config,
@@ -68,6 +72,7 @@ impl Default for Config {
             allow_unsynced_beacon_node: false,
             http_api: <_>::default(),
             // http_metrics: <_>::default(),
+            network: <_>::default(),
             beacon_nodes_tls_certs: None,
             execution_nodes_tls_certs: None,
         }
@@ -119,6 +124,11 @@ pub fn from_cli(cli_args: &Anchor) -> Result<Config, String> {
             .map_err(|e| format!("Unable to parse execution node URL: {:?}", e))?;
     }
 
+    /*
+     * Network related
+     */
+    config.network.listen_addresses = parse_listening_addresses(cli_args)?;
+
     config.beacon_nodes_tls_certs = cli_args.beacon_nodes_tls_certs.clone();
     config.execution_nodes_tls_certs = cli_args.execution_nodes_tls_certs.clone();
 
@@ -190,6 +200,182 @@ pub fn from_cli(cli_args: &Anchor) -> Result<Config, String> {
     Ok(config)
 }
 
+/// Gets the listening_addresses for lighthouse based on the cli options.
+pub fn parse_listening_addresses(cli_args: &Anchor) -> Result<ListenAddress, String> {
+    // parse the possible ips
+    let mut maybe_ipv4 = None;
+    let mut maybe_ipv6 = None;
+    for addr in cli_args.listen_addresses.iter() {
+        match addr {
+            IpAddr::V4(v4_addr) => match &maybe_ipv4 {
+                Some(first_ipv4_addr) => {
+                    return Err(format!(
+                        "When setting the --listen-address option twice, use an IpV4 address and an Ipv6 address. \
+                                Got two IpV4 addresses {first_ipv4_addr} and {v4_addr}"
+                    ));
+                }
+                None => maybe_ipv4 = Some(v4_addr),
+            },
+            IpAddr::V6(v6_addr) => match &maybe_ipv6 {
+                Some(first_ipv6_addr) => {
+                    return Err(format!(
+                        "When setting the --listen-address option twice, use an IpV4 address and an Ipv6 address. \
+                                Got two IpV6 addresses {first_ipv6_addr} and {v6_addr}"
+                    ));
+                }
+                None => maybe_ipv6 = Some(v6_addr),
+            },
+        }
+    }
+
+    // Now put everything together
+    let listening_addresses = match (maybe_ipv4, maybe_ipv6) {
+        (None, None) => {
+            // This should never happen unless clap is broken
+            return Err("No listening addresses provided".into());
+        }
+        (None, Some(ipv6)) => {
+            // A single ipv6 address was provided. Set the ports
+            if cli_args.port6.is_some() {
+                warn!("When listening only over IPv6, use the --port flag. The value of --port6 will be ignored.");
+            }
+
+            if cli_args.discovery_port6.is_some() {
+                warn!("When listening only over IPv6, use the --discovery-port flag. The value of --discovery-port6 will be ignored.")
+            }
+
+            if cli_args.quic_port6.is_some() {
+                warn!("When listening only over IPv6, use the --quic-port flag. The value of --quic-port6 will be ignored.")
+            }
+
+            // use zero ports if required. If not, use the given port.
+            let tcp_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_tcp6_port)
+                .transpose()?
+                .unwrap_or(cli_args.port);
+
+            // use zero ports if required. If not, use the specific udp port. If none given, use
+            // the tcp port.
+            let disc_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp6_port)
+                .transpose()?
+                .or(cli_args.discovery_port)
+                .unwrap_or(tcp_port);
+
+            let quic_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp6_port)
+                .transpose()?
+                .or(cli_args.quic_port)
+                .unwrap_or(if tcp_port == 0 { 0 } else { tcp_port + 1 });
+
+            ListenAddress::V6(ListenAddr {
+                addr: *ipv6,
+                quic_port,
+                disc_port,
+                tcp_port,
+            })
+        }
+        (Some(ipv4), None) => {
+            // A single ipv4 address was provided. Set the ports
+
+            // use zero ports if required. If not, use the given port.
+            let tcp_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_tcp4_port)
+                .transpose()?
+                .unwrap_or(cli_args.port);
+            // use zero ports if required. If not, use the specific discovery port. If none given, use
+            // the tcp port.
+            let disc_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp4_port)
+                .transpose()?
+                .or(cli_args.discovery_port)
+                .unwrap_or(tcp_port);
+            // use zero ports if required. If not, use the specific quic port. If none given, use
+            // the tcp port + 1.
+            let quic_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp4_port)
+                .transpose()?
+                .or(cli_args.quic_port)
+                .unwrap_or(if tcp_port == 0 { 0 } else { tcp_port + 1 });
+
+            ListenAddress::V4(ListenAddr {
+                addr: *ipv4,
+                disc_port,
+                quic_port,
+                tcp_port,
+            })
+        }
+        (Some(ipv4), Some(ipv6)) => {
+            let ipv4_tcp_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_tcp4_port)
+                .transpose()?
+                .unwrap_or(cli_args.port);
+            let ipv4_disc_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp4_port)
+                .transpose()?
+                .or(cli_args.discovery_port)
+                .unwrap_or(ipv4_tcp_port);
+            let ipv4_quic_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp4_port)
+                .transpose()?
+                .or(cli_args.quic_port)
+                .unwrap_or(if ipv4_tcp_port == 0 {
+                    0
+                } else {
+                    ipv4_tcp_port + 1
+                });
+
+            let ipv6_tcp_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_tcp6_port)
+                .transpose()?
+                .unwrap_or(cli_args.port);
+            let ipv6_disc_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp6_port)
+                .transpose()?
+                .or(cli_args.discovery_port6)
+                .unwrap_or(ipv6_tcp_port);
+            let ipv6_quic_port = cli_args
+                .use_zero_ports
+                .then(unused_port::unused_udp6_port)
+                .transpose()?
+                .or(cli_args.quic_port6)
+                .unwrap_or(if ipv6_tcp_port == 0 {
+                    0
+                } else {
+                    ipv6_tcp_port + 1
+                });
+
+            ListenAddress::DualStack(
+                ListenAddr {
+                    addr: *ipv4,
+                    disc_port: ipv4_disc_port,
+                    quic_port: ipv4_quic_port,
+                    tcp_port: ipv4_tcp_port,
+                },
+                ListenAddr {
+                    addr: *ipv6,
+                    disc_port: ipv6_disc_port,
+                    quic_port: ipv6_quic_port,
+                    tcp_port: ipv6_tcp_port,
+                },
+            )
+        }
+    };
+
+    Ok(listening_addresses)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

anchor/client/src/lib.rs

@@ -2,18 +2,18 @@
 
 mod cli;
 pub mod config;
-mod version;
 
 pub use cli::Anchor;
 use config::Config;
+use network::Network;
 use task_executor::TaskExecutor;
 use tracing::{debug, error, info};
 
 pub struct Client {}
 
 impl Client {
     /// Runs the Anchor Client
-    pub async fn run(_executor: TaskExecutor, config: Config) -> Result<(), String> {
+    pub async fn run(executor: TaskExecutor, config: Config) -> Result<(), String> {
         // Attempt to raise soft fd limit. The behavior is OS specific:
         // `linux` - raise soft fd limit to hard
         // `macos` - raise soft fd limit to `min(kernel limit, hard fd limit)`
@@ -76,6 +76,10 @@ impl Client {
             error!(error, "Failed to run HTTP API");
             return Err("HTTP API Failed".to_string());
         }
+
+        info!("Starting the network service");
+        Network::spawn(executor, &config.network);
+
         Ok(())
     }
 }

anchor/common/version/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "version"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+git-version = "0.3.9"
+target_info = "0.1.0"
+
+[dev-dependencies]
+regex = "1.11"