Consensus Integration (#1467)

narwhal integration

deny.toml

@@ -143,7 +143,7 @@ exceptions = [
 name = "ring"
 expression = "LicenseRef-ring"
 license-files = [
-    { path = "LICENSE", hash = 0xbd0eed23 },
+    { path = "LICENSE", hash = 0xbd0eed23 }
 ]
 [[licenses.clarify]]
 name = "encoding_rs"
@@ -152,6 +152,12 @@ expression = "(Apache-2.0 OR MIT) AND BSD-3-Clause"
 license-files = [
     { path = "COPYRIGHT", hash = 0x39f8ad31 }
 ]
+[[licenses.clarify]]
+name = "bls-crypto"
+version = "*"
+expression = "MIT AND Apache-2.0"
+license-files = [
+]
 
 [licenses.private]
 # If true, ignores workspace crates that aren't published, or are only

sui/Cargo.toml

@@ -60,6 +60,10 @@ move-binary-format = { git = "https://github.com/move-language/move", rev = "6a8
 move-bytecode-utils = { git = "https://github.com/move-language/move", rev = "6a80792ecbf16d74bf1d57e48a576377f0879646" }
 move-unit-test = { git = "https://github.com/move-language/move", rev = "6a80792ecbf16d74bf1d57e48a576377f0879646" }
 
+narwhal-node = { git = "https://github.com/MystenLabs/narwhal", rev = "aad875a157ba1897b7f277dae7d9cb1deef7955c", package = "node" }
+narwhal-config = { git = "https://github.com/MystenLabs/narwhal", rev = "aad875a157ba1897b7f277dae7d9cb1deef7955c", package = "config" }
+narwhal-crypto = { git = "https://github.com/MystenLabs/narwhal", rev = "aad875a157ba1897b7f277dae7d9cb1deef7955c", package = "crypto" }
+
 once_cell = "1.9.0"
 reqwest = { version = "0.11.10", features = ["json", "serde_json", "blocking"] }
 

sui/src/benchmark/load_generator.rs

@@ -176,7 +176,7 @@ pub async fn spawn_authority_server(
         network_server.base_address,
         network_server.base_port,
         network_server.buffer_size,
-        state,
+        Arc::new(state),
         consensus_address,
         tx_consensus_listener,
     );

sui/src/config.rs

@@ -2,35 +2,37 @@
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::fmt::Write;
-use std::fmt::{Display, Formatter};
-use std::fs::{self, File};
-use std::io::BufReader;
-use std::ops::{Deref, DerefMut};
-use std::path::{Path, PathBuf};
-use std::sync::Mutex;
-
+use crate::gateway::GatewayType;
+use crate::keystore::KeystoreType;
+use narwhal_config::Committee as ConsensusCommittee;
+use narwhal_config::{Authority, PrimaryAddresses, Stake, WorkerAddresses};
+use narwhal_crypto::ed25519::Ed25519PublicKey;
 use once_cell::sync::Lazy;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use serde_with::hex::Hex;
 use serde_with::serde_as;
-use sui_types::committee::Committee;
-use tracing::log::trace;
-
-use crate::gateway::GatewayType;
-use crate::keystore::KeystoreType;
+use std::fmt::Write;
+use std::fmt::{Display, Formatter};
+use std::fs::{self, File};
+use std::io::BufReader;
 use std::net::SocketAddr;
+use std::ops::{Deref, DerefMut};
+use std::path::{Path, PathBuf};
+use std::sync::Mutex;
 use sui_framework::DEFAULT_FRAMEWORK_PATH;
 use sui_network::network::PortAllocator;
 use sui_types::base_types::*;
+use sui_types::committee::Committee;
 use sui_types::crypto::{get_key_pair, KeyPair};
+use tracing::log::trace;
 
 const DEFAULT_WEIGHT: usize = 1;
 const DEFAULT_GAS_AMOUNT: u64 = 100000;
 pub const AUTHORITIES_DB_NAME: &str = "authorities_db";
 pub const DEFAULT_STARTING_PORT: u16 = 10000;
+pub const CONSENSUS_DB_NAME: &str = "consensus_db";
 
 static PORT_ALLOCATOR: Lazy<Mutex<PortAllocator>> =
     Lazy::new(|| Mutex::new(PortAllocator::new(DEFAULT_STARTING_PORT)));
@@ -326,3 +328,55 @@ impl<C> DerefMut for PersistedConfig<C> {
         &mut self.inner
     }
 }
+
+/// Make a default Narwhal-compatible committee.
+pub fn make_default_narwhal_committee(
+    authorities: &[AuthorityPrivateInfo],
+) -> Result<ConsensusCommittee<Ed25519PublicKey>, anyhow::Error> {
+    let mut ports = Vec::new();
+    for _ in authorities {
+        let mut authority_ports = Vec::new();
+        for _ in 0..4 {
+            let port = PORT_ALLOCATOR
+                .lock()
+                .map_err(|e| anyhow::anyhow!("{e}"))?
+                .next_port()
+                .ok_or_else(|| anyhow::anyhow!("No available ports"))?;
+            authority_ports.push(port);
+        }
+        ports.push(authority_ports);
+    }
+
+    Ok(ConsensusCommittee {
+        authorities: authorities
+            .iter()
+            .enumerate()
+            .map(|(i, x)| {
+                let name = x.key_pair.make_narwhal_keypair().name;
+
+                let primary = PrimaryAddresses {
+                    primary_to_primary: format!("127.0.0.1:{}", ports[i][0]).parse().unwrap(),
+                    worker_to_primary: format!("127.0.0.1:{}", ports[i][1]).parse().unwrap(),
+                };
+                let workers = [(
+                    /* worker_id */ 0,
+                    WorkerAddresses {
+                        primary_to_worker: format!("127.0.0.1:{}", ports[i][2]).parse().unwrap(),
+                        transactions: x.consensus_address,
+                        worker_to_worker: format!("127.0.0.1:{}", ports[i][3]).parse().unwrap(),
+                    },
+                )]
+                .iter()
+                .cloned()
+                .collect();
+
+                let authority = Authority {
+                    stake: x.stake as Stake,
+                    primary,
+                    workers,
+                };
+                (name, authority)
+            })
+            .collect(),
+    })
+}

sui/src/sui_commands.rs

@@ -1,14 +1,23 @@
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-
+use crate::config::{make_default_narwhal_committee, CONSENSUS_DB_NAME};
+use crate::config::{
+    AuthorityPrivateInfo, Config, GenesisConfig, NetworkConfig, PersistedConfig, WalletConfig,
+};
+use crate::gateway::{GatewayConfig, GatewayType};
+use crate::keystore::{Keystore, KeystoreType, SuiKeystore};
+use crate::{sui_config_dir, SUI_GATEWAY_CONFIG, SUI_NETWORK_CONFIG, SUI_WALLET_CONFIG};
 use anyhow::{anyhow, bail};
 use base64ct::{Base64, Encoding};
 use clap::*;
 use futures::future::join_all;
 use move_binary_format::CompiledModule;
 use move_package::BuildConfig;
+use narwhal_config::{Committee as ConsensusCommittee, Parameters as ConsensusParameters};
+use narwhal_crypto::ed25519::Ed25519PublicKey;
 use std::collections::BTreeMap;
 use std::fs;
+use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
 use sui_adapter::adapter::generate_package_id;
@@ -19,20 +28,14 @@ use sui_core::consensus_adapter::ConsensusListener;
 use sui_network::transport::SpawnedServer;
 use sui_network::transport::DEFAULT_MAX_DATAGRAM_SIZE;
 use sui_types::base_types::decode_bytes_hex;
+use sui_types::base_types::encode_bytes_hex;
 use sui_types::base_types::{SequenceNumber, SuiAddress, TxContext};
 use sui_types::committee::Committee;
 use sui_types::error::SuiResult;
 use sui_types::object::Object;
 use tokio::sync::mpsc::channel;
 use tracing::{error, info};
 
-use crate::config::{
-    AuthorityPrivateInfo, Config, GenesisConfig, NetworkConfig, PersistedConfig, WalletConfig,
-};
-use crate::gateway::{GatewayConfig, GatewayType};
-use crate::keystore::{Keystore, KeystoreType, SuiKeystore};
-use crate::{sui_config_dir, SUI_GATEWAY_CONFIG, SUI_NETWORK_CONFIG, SUI_WALLET_CONFIG};
-
 #[derive(Parser)]
 #[clap(rename_all = "kebab-case")]
 pub enum SuiCommand {
@@ -78,6 +81,7 @@ impl SuiCommand {
     pub async fn execute(&self) -> Result<(), anyhow::Error> {
         match self {
             SuiCommand::Start { config } => {
+                // Load the config of the Sui authority.
                 let config_path = config
                     .clone()
                     .unwrap_or(sui_config_dir()?.join(SUI_NETWORK_CONFIG));
@@ -87,6 +91,8 @@ impl SuiCommand {
                         config_path
                     ))
                 })?;
+
+                // Start a sui validator (including its consensus node).
                 SuiNetwork::start(&config)
                     .await?
                     .wait_for_completion()
@@ -277,9 +283,24 @@ impl SuiNetwork {
                 .collect(),
         );
 
+        let consensus_committee = make_default_narwhal_committee(&config.authorities)?;
+        let consensus_parameters = ConsensusParameters::default();
+
         let mut spawned_authorities = Vec::new();
         for authority in &config.authorities {
-            let server = make_server(authority, &committee, config.buffer_size).await?;
+            let consensus_store_path = sui_config_dir()?
+                .join(CONSENSUS_DB_NAME)
+                .join(encode_bytes_hex(authority.key_pair.public_key_bytes()));
+
+            let server = make_server(
+                authority,
+                &committee,
+                config.buffer_size,
+                &consensus_committee,
+                &consensus_store_path,
+                &consensus_parameters,
+            )
+            .await?;
             spawned_authorities.push(server.spawn().await?);
         }
         info!("Started {} authorities", spawned_authorities.len());
@@ -325,7 +346,6 @@ pub async fn genesis(
         loaded_move_packages: vec![],
     };
     let mut voting_right = BTreeMap::new();
-
     for authority in genesis_conf.authorities {
         voting_right.insert(*authority.key_pair.public_key_bytes(), authority.stake);
         network_config.authorities.push(authority);
@@ -420,6 +440,9 @@ pub async fn make_server(
     authority: &AuthorityPrivateInfo,
     committee: &Committee,
     buffer_size: usize,
+    consensus_committee: &ConsensusCommittee<Ed25519PublicKey>,
+    consensus_store_path: &Path,
+    consensus_parameters: &ConsensusParameters,
 ) -> SuiResult<AuthorityServer> {
     let store = Arc::new(AuthorityStore::open(&authority.db_path, None));
     let name = *authority.key_pair.public_key_bytes();
@@ -431,7 +454,15 @@ pub async fn make_server(
     )
     .await;
 
-    make_authority(authority, buffer_size, state).await
+    make_authority(
+        authority,
+        buffer_size,
+        state,
+        consensus_committee,
+        consensus_store_path,
+        consensus_parameters,
+    )
+    .await
 }
 
 async fn make_server_with_genesis_ctx(
@@ -459,7 +490,15 @@ async fn make_server_with_genesis_ctx(
         state.insert_genesis_object(object.clone()).await;
     }
 
-    make_authority(authority, buffer_size, state).await
+    let (tx_sui_to_consensus, _rx_sui_to_consensus) = channel(1);
+    Ok(AuthorityServer::new(
+        authority.host.clone(),
+        authority.port,
+        buffer_size,
+        Arc::new(state),
+        authority.consensus_address,
+        /* tx_consensus_listener */ tx_sui_to_consensus,
+    ))
 }
 
 /// Spawn all the subsystems run by a Sui authority: a consensus node, a sui authority server,
@@ -468,12 +507,36 @@ async fn make_authority(
     authority: &AuthorityPrivateInfo,
     buffer_size: usize,
     state: AuthorityState,
+    consensus_committee: &ConsensusCommittee<Ed25519PublicKey>,
+    consensus_store_path: &Path,
+    consensus_parameters: &ConsensusParameters,
 ) -> SuiResult<AuthorityServer> {
     let (tx_consensus_to_sui, rx_consensus_to_sui) = channel(1_000);
     let (tx_sui_to_consensus, rx_sui_to_consensus) = channel(1_000);
 
-    // TODO [issue #633]: Spawn the consensus node of this authority.
-    let _tx_consensus_to_sui = tx_consensus_to_sui;
+    let authority_state = Arc::new(state);
+
+    // Spawn the consensus node of this authority.
+    let consensus_keypair = authority.key_pair.make_narwhal_keypair();
+    let consensus_name = consensus_keypair.name.clone();
+    let consensus_store = narwhal_node::NodeStorage::reopen(consensus_store_path);
+    narwhal_node::Node::spawn_primary(
+        consensus_keypair,
+        consensus_committee.clone(),
+        &consensus_store,
+        consensus_parameters.clone(),
+        /* consensus */ true, // Indicate that we want to run consensus.
+        /* execution_state */ authority_state.clone(),
+        /* tx_confirmation */ tx_consensus_to_sui,
+    )
+    .await?;
+    narwhal_node::Node::spawn_workers(
+        consensus_name,
+        /* ids */ vec![0], // We run a single worker with id '0'.
+        consensus_committee.clone(),
+        &consensus_store,
+        consensus_parameters.clone(),
+    );
 
     // Spawn a consensus listener. It listen for consensus outputs and notifies the
     // authority server when a sequenced transaction is ready for execution.
@@ -484,7 +547,7 @@ async fn make_authority(
         authority.host.clone(),
         authority.port,
         buffer_size,
-        state,
+        authority_state,
         authority.consensus_address,
         /* tx_consensus_listener */ tx_sui_to_consensus,
     ))

sui/src/validator.rs

@@ -3,12 +3,17 @@
 
 use anyhow::anyhow;
 use clap::*;
+use narwhal_config::Parameters as ConsensusParameters;
 use std::path::PathBuf;
+use sui::config::make_default_narwhal_committee;
+use sui::config::CONSENSUS_DB_NAME;
+use sui::sui_config_dir;
 use sui::{
     config::{GenesisConfig, NetworkConfig, PersistedConfig},
     sui_commands::{genesis, make_server},
     sui_config_dir, SUI_NETWORK_CONFIG,
 };
+use sui_types::base_types::encode_bytes_hex;
 use sui_types::base_types::{decode_bytes_hex, SuiAddress};
 use sui_types::committee::Committee;
 use tracing::{error, info};
@@ -97,10 +102,19 @@ async fn main() -> Result<(), anyhow::Error> {
         net_cfg.port
     );
 
+    let consensus_committee = make_default_narwhal_committee(&network_config.authorities)?;
+    let consensus_parameters = ConsensusParameters::default();
+    let consensus_store_path = sui_config_dir()?
+        .join(CONSENSUS_DB_NAME)
+        .join(encode_bytes_hex(net_cfg.key_pair.public_key_bytes()));
+
     if let Err(e) = make_server(
         net_cfg,
         &Committee::from(&network_config),
         network_config.buffer_size,
+        &consensus_committee,
+        &consensus_store_path,
+        &consensus_parameters,
     )
     .await
     .unwrap()

sui_core/Cargo.toml

@@ -17,7 +17,7 @@ serde = { version = "1.0.136", features = ["derive"] }
 tokio = { version = "1.17.0", features = ["full", "tracing"] }
 parking_lot = "0.12.0"
 itertools = "0.10.3"
-async-trait = "0.1.52"
+async-trait = "0.1.53"
 tempfile = "3.3.0"
 tracing = { version = "0.1.31", features = ["log"] }
 signature = "1.5.0"
@@ -39,6 +39,8 @@ move-vm-types = { git = "https://github.com/move-language/move", rev = "6a80792e
 
 typed-store = { git = "https://github.com/MystenLabs/mysten-infra", rev = "e44bca4513a6ff6c97399cd79e82e4bc00571ac3"}
 
+narwhal-executor = { git = "https://github.com/MystenLabs/narwhal", rev = "aad875a157ba1897b7f277dae7d9cb1deef7955c", package = "executor" }
+
 [dev-dependencies]
 fdlimit = "0.2.1"
 serde-reflection = "0.3.5"