Add basic node CLI with support for 3 peers

code/Cargo.toml

@@ -29,24 +29,26 @@ malachite-round  = { version = "0.1.0", path = "round" }
 malachite-test   = { version = "0.1.0", path = "test" }
 malachite-vote   = { version = "0.1.0", path = "vote" }
 
-derive-where      = "1.2.7"
-ed25519-consensus = "2.1.0"
-futures           = "0.3"
-glob              = "0.3.0"
-itf               = "0.2.2"
-num-bigint        = "0.4.4"
-num-traits        = "0.2.17"
-pretty_assertions = "1.4"
-prost             = "0.12.3"
-prost-types       = "0.12.3"
-prost-build       = "0.12.3"
-rand              = { version = "0.8.5", features = ["std_rng"] }
-serde             = "1.0"
-serde_json        = "1.0"
-serde_with        = "3.4"
-sha2              = "0.10.8"
-signature         = "2.1.0"
-thiserror         = "1.0"
-tokio             = "1.35.1"
-tokio-stream      = "0.1"
-tracing           = "0.1.40"
+derive-where       = "1.2.7"
+ed25519-consensus  = "2.1.0"
+futures            = "0.3"
+glob               = "0.3.0"
+itf                = "0.2.2"
+num-bigint         = "0.4.4"
+num-traits         = "0.2.17"
+pretty_assertions  = "1.4"
+prost              = "0.12.3"
+prost-types        = "0.12.3"
+prost-build        = "0.12.3"
+rand               = "0.8.5"
+serde              = "1.0"
+serde_json         = "1.0"
+serde_with         = "3.4"
+sha2               = "0.10.8"
+signature          = "2.1.0"
+thiserror          = "1.0"
+tokio              = "1.35.1"
+tokio-stream       = "0.1"
+toml               = "0.8.10"
+tracing            = "0.1.40"
+tracing-subscriber = "0.3.18"

code/common/src/timeout.rs

@@ -1,3 +1,5 @@
+use core::fmt;
+
 use crate::Round;
 
 /// The round step for which the timeout is for.
@@ -44,3 +46,9 @@ impl Timeout {
         Self::new(round, TimeoutStep::Precommit)
     }
 }
+
+impl fmt::Display for Timeout {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{:?}Timeout({})", self.step, self.round)
+    }
+}

code/node/Cargo.toml

@@ -8,20 +8,29 @@ repository.workspace = true
 license.workspace    = true
 publish.workspace    = true
 
+[[bin]]
+name = "malachite-node"
+path = "bin/main.rs"
+
 [dependencies]
 malachite-common.workspace = true
 malachite-driver.workspace = true
 malachite-round.workspace  = true
 malachite-vote.workspace   = true
 malachite-proto.workspace  = true
+malachite-test.workspace   = true
 
-derive-where = { workspace = true }
-futures      = { workspace = true }
-tokio        = { workspace = true, features = ["full"] }
-tokio-stream = { workspace = true }
-prost        = { workspace = true }
-prost-types  = { workspace = true }
-tracing      = { workspace = true }
+ed25519-consensus  = { workspace = true, features = ["serde"] }
+derive-where       = { workspace = true }
+futures            = { workspace = true }
+tokio              = { workspace = true, features = ["full"] }
+tokio-stream       = { workspace = true }
+prost              = { workspace = true }
+prost-types        = { workspace = true }
+serde              = { workspace = true, features = ["derive"] }
+toml               = { workspace = true }
+tracing            = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["fmt"] }
 
 [dev-dependencies]
 malachite-test.workspace = true

code/node/bin/cli.rs

@@ -0,0 +1,17 @@
+use malachite_node::network::PeerId;
+
+pub struct Cli {
+    pub peer_id: PeerId,
+}
+
+impl Cli {
+    pub fn from_env() -> Self {
+        let peer_id = std::env::args()
+            .nth(1)
+            .expect("Usage: node <PEER_ID>")
+            .parse()
+            .expect("Error: Invalid PEER_ID");
+
+        Self { peer_id }
+    }
+}

code/node/bin/config.rs

@@ -0,0 +1,74 @@
+use std::net::SocketAddr;
+
+use serde::{Deserialize, Serialize};
+
+use malachite_node::network::{broadcast::PeerInfo, PeerId};
+use malachite_test::PublicKey;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Config {
+    pub peers: Vec<PeerConfig>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct PeerConfig {
+    #[serde(with = "de::peer_id")]
+    pub id: PeerId,
+    pub addr: SocketAddr,
+    #[serde(with = "de::public_key")]
+    pub public_key: PublicKey,
+}
+
+impl PeerConfig {
+    pub fn peer_info(&self) -> PeerInfo {
+        PeerInfo {
+            id: self.id.clone(),
+            addr: self.addr,
+        }
+    }
+}
+
+pub mod de {
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+    pub mod peer_id {
+        use super::*;
+
+        use malachite_node::network::PeerId;
+
+        pub fn serialize<S>(id: &PeerId, s: S) -> Result<S::Ok, S::Error>
+        where
+            S: Serializer,
+        {
+            s.serialize_str(&id.to_string())
+        }
+
+        pub fn deserialize<'de, D>(d: D) -> Result<PeerId, D::Error>
+        where
+            D: Deserializer<'de>,
+        {
+            let s = String::deserialize(d)?;
+            Ok(PeerId::new(s))
+        }
+    }
+
+    pub mod public_key {
+        use super::*;
+
+        use malachite_test::PublicKey;
+
+        pub fn serialize<S>(key: &PublicKey, s: S) -> Result<S::Ok, S::Error>
+        where
+            S: Serializer,
+        {
+            key.inner().serialize(s)
+        }
+
+        pub fn deserialize<'de, D>(d: D) -> Result<PublicKey, D::Error>
+        where
+            D: Deserializer<'de>,
+        {
+            ed25519_consensus::VerificationKey::deserialize(d).map(PublicKey::new)
+        }
+    }
+}

code/node/bin/main.rs

@@ -0,0 +1,93 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use malachite_node::network::broadcast;
+use malachite_node::network::broadcast::PeerInfo;
+use malachite_node::node::{Node, Params};
+use malachite_node::timers;
+use malachite_test::utils::{make_validators, RotateProposer};
+use malachite_test::{Address, Height, PrivateKey, TestContext, ValidatorSet};
+use tracing::info;
+
+mod cli;
+use cli::Cli;
+mod config;
+use config::{Config, PeerConfig};
+
+#[tokio::main(flavor = "current_thread")]
+pub async fn main() {
+    tracing_subscriber::fmt::init();
+
+    let args = Cli::from_env();
+
+    // Validators keys are deterministic and match the ones in the config file
+    let vs = make_validators([2, 3, 2]);
+
+    let config = std::fs::read_to_string("node/peers.toml").expect("Error: missing peers.toml");
+    let config = toml::from_str::<Config>(&config).expect("Error: invalid peers.toml");
+
+    let peer_config = config
+        .peers
+        .iter()
+        .find(|p| p.id == args.peer_id)
+        .expect("Error: invalid peer id");
+
+    let (my_sk, my_addr) = vs
+        .iter()
+        .find(|(v, _)| v.public_key == peer_config.public_key)
+        .map(|(v, pk)| (pk.clone(), v.address))
+        .expect("Error: invalid peer id");
+
+    let (vs, _): (Vec<_>, Vec<_>) = vs.into_iter().unzip();
+
+    let peer_info = peer_config.peer_info();
+    let vs = ValidatorSet::new(vs);
+
+    let node = make_node(vs, my_sk, my_addr, peer_info, &config.peers).await;
+
+    info!("[{}] Starting...", args.peer_id);
+
+    node.run().await;
+}
+
+pub async fn make_node(
+    vs: ValidatorSet,
+    pk: PrivateKey,
+    addr: Address,
+    peer_info: PeerInfo,
+    peers: &[PeerConfig],
+) -> Node<TestContext, broadcast::Handle> {
+    let height = Height::new(1);
+    let ctx = TestContext::new(pk);
+    let sel = Arc::new(RotateProposer);
+
+    let params = Params {
+        start_height: height,
+        proposer_selector: sel,
+        validator_set: vs,
+        address: addr,
+        threshold_params: Default::default(),
+    };
+
+    let timers_config = timers::Config {
+        propose_timeout: Duration::from_secs(3),
+        prevote_timeout: Duration::from_secs(1),
+        precommit_timeout: Duration::from_secs(1),
+    };
+
+    let network = broadcast::Peer::new(peer_info.clone());
+    let handle = network.run().await;
+
+    let timeout = Some(Duration::from_secs(10));
+
+    let to_connect = peers
+        .iter()
+        .filter(|p| p.id != peer_info.id)
+        .map(|p| p.peer_info());
+
+    for peer in to_connect {
+        handle.connect_to_peer(peer, timeout).await;
+    }
+
+    Node::new(ctx, params, handle, timers_config)
+}

code/node/peers.toml

@@ -0,0 +1,14 @@
+[[peers]]
+id = "node1"
+addr = "127.0.0.1:1235"
+public_key = [104, 30, 170, 163, 78, 73, 30, 108, 131, 53, 171, 201, 234, 146, 176, 36, 239, 82, 235, 145, 68, 44, 163, 184, 69, 152, 199, 154, 121, 243, 27, 117]
+
+[[peers]]
+id = "node2"
+addr = "127.0.0.1:1236"
+public_key = [24, 109, 62, 237, 160, 46, 173, 95, 187, 173, 116, 78, 237, 21, 141, 149, 140, 79, 127, 72, 86, 26, 62, 102, 203, 30, 233, 104, 85, 173, 92, 25]
+
+[[peers]]
+id = "node3"
+addr = "127.0.0.1:1237"
+public_key = [49, 171, 95, 10, 5, 226, 72, 164, 147, 3, 48, 71, 200, 88, 31, 0, 121, 180, 85, 143, 94, 156, 113, 175, 97, 106, 231, 109, 128, 203, 219, 7]

code/node/src/network.rs

@@ -1,4 +1,6 @@
 use core::fmt;
+use std::convert::Infallible;
+use std::str::FromStr;
 
 pub mod broadcast;
 mod msg;
@@ -10,12 +12,26 @@ pub use self::msg::Msg;
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct PeerId(String);
 
+impl PeerId {
+    pub fn new(id: impl ToString) -> Self {
+        Self(id.to_string())
+    }
+}
+
 impl fmt::Display for PeerId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt::Display::fmt(&self.0, f)
     }
 }
 
+impl FromStr for PeerId {
+    type Err = Infallible;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(Self(s.to_string()))
+    }
+}
+
 #[allow(async_fn_in_trait)]
 pub trait Network {
     async fn recv(&mut self) -> Option<(PeerId, Msg)>;