feat: add validator node registration (#4507)

applications/tari_app_grpc/proto/base_node.proto

@@ -88,6 +88,9 @@ service BaseNode {
     rpc ListConnectedPeers(Empty) returns (ListConnectedPeersResponse);
     // Get mempool stats
     rpc GetMempoolStats(Empty) returns (MempoolStatsResponse);
+    // Get VNs
+    rpc GetActiveValidatorNodes(GetActiveValidatorNodesRequest) returns (stream ActiveValidatorNode);
+    rpc GetCommittee(GetCommitteeRequest) returns (GetCommitteeResponse);
 
 }
 
@@ -434,3 +437,22 @@ message MempoolStatsResponse {
     uint64 unconfirmed_weight = 4;
 }
 
+message GetActiveValidatorNodesRequest {
+    uint64 height = 1;
+}
+
+message ActiveValidatorNode {
+    bytes shard_key = 1;
+    uint64 from_height = 2;
+    uint64 to_height = 3;
+    bytes public_key = 4;
+}
+
+message GetCommitteeRequest {
+    uint64 height = 1;
+    bytes shard_key = 2;
+}
+
+message GetCommitteeResponse {
+    repeated bytes public_key = 1;
+}

applications/tari_app_grpc/proto/block.proto

@@ -61,6 +61,8 @@ message BlockHeader {
     uint64 output_mmr_size = 14;
     // Sum of script offsets for all kernels in this block.
     bytes total_script_offset = 15;
+    // Merkle root of validator nodes
+    bytes validator_node_merkle_root = 16;
 }
 
 // Metadata required for validating the Proof of Work calculation
@@ -117,6 +119,8 @@ message NewBlockHeaderTemplate {
 //    uint64 target_difficulty = 6;
     // Sum of script offsets for all kernels in this block.
     bytes total_script_offset = 7;
+    // Merkle root of validator nodes
+    bytes validator_node_merkle_root = 8;
 }
 
 // The new block template is used constructing a new partial block, allowing a miner to added the coinbase utxo and as a final step the Base node to add the MMR roots to the header.

applications/tari_app_grpc/proto/transaction.proto

@@ -124,6 +124,8 @@ message OutputFeatures {
     uint64 maturity = 3;
     bytes metadata = 4;
     SideChainFeatures sidechain_features = 5;
+    bytes validator_node_public_key = 6;
+    Signature validator_node_signature = 7;
 }
 
 

applications/tari_app_grpc/proto/wallet.proto

@@ -78,6 +78,8 @@ service Wallet {
     rpc StreamTransactionEvents(TransactionEventRequest) returns (stream TransactionEventResponse);
     rpc SeedWords(Empty) returns (SeedWordsResponse);
     rpc DeleteSeedWordsFile(Empty) returns (FileDeletedResponse);
+
+    rpc RegisterValidatorNode(RegisterValidatorNodeRequest) returns (RegisterValidatorNodeResponse);
 }
 
 message GetVersionRequest { }
@@ -326,4 +328,17 @@ message SeedWordsResponse {
 
 message FileDeletedResponse {
 
+}
+
+message RegisterValidatorNodeRequest {
+    string validator_node_public_key = 1;
+    Signature validator_node_signature = 2;
+    uint64 fee_per_gram = 3;
+    string message = 4;
+}
+
+message RegisterValidatorNodeResponse {
+    uint64 transaction_id = 1;
+    bool is_success = 2;
+    string failure_message = 3;
 }

applications/tari_app_grpc/src/conversions/active_validator_node.rs

@@ -0,0 +1,59 @@
+// Copyright 2020. The Tari Project
+//
+// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+// following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+// disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+// following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+// products derived from this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use std::convert::{TryFrom, TryInto};
+
+use tari_common_types::types::PublicKey;
+use tari_core::chain_storage::ActiveValidatorNode;
+use tari_utilities::ByteArray;
+
+use crate::tari_rpc as grpc;
+
+impl TryFrom<tari_core::chain_storage::ActiveValidatorNode> for grpc::ActiveValidatorNode {
+    type Error = String;
+
+    fn try_from(active_validator_node: ActiveValidatorNode) -> Result<Self, Self::Error> {
+        Ok(Self {
+            shard_key: active_validator_node.shard_key.to_vec(),
+            from_height: active_validator_node.from_height,
+            to_height: active_validator_node.to_height,
+            public_key: active_validator_node.public_key.to_vec(),
+        })
+    }
+}
+
+impl TryFrom<grpc::ActiveValidatorNode> for ActiveValidatorNode {
+    type Error = String;
+
+    fn try_from(active_validator_node: grpc::ActiveValidatorNode) -> Result<Self, Self::Error> {
+        let shard_key = active_validator_node.shard_key.try_into().unwrap();
+        let public_key =
+            PublicKey::from_vec(&active_validator_node.public_key).map_err(|_| "Could not public key".to_string())?;
+
+        Ok(Self {
+            shard_key,
+            from_height: active_validator_node.from_height,
+            to_height: active_validator_node.to_height,
+            public_key,
+        })
+    }
+}

applications/tari_app_grpc/src/conversions/block_header.rs

@@ -53,6 +53,7 @@ impl From<BlockHeader> for grpc::BlockHeader {
                 pow_algo: pow_algo.as_u64(),
                 pow_data: h.pow.pow_data,
             }),
+            validator_node_merkle_root: h.validator_node_merkle_root,
         }
     }
 }
@@ -91,6 +92,7 @@ impl TryFrom<grpc::BlockHeader> for BlockHeader {
             total_script_offset,
             nonce: header.nonce,
             pow,
+            validator_node_merkle_root: header.validator_node_merkle_root,
         })
     }
 }

applications/tari_app_grpc/src/conversions/mod.rs

@@ -20,6 +20,7 @@
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+mod active_validator_node;
 mod aggregate_body;
 mod base_node_state;
 mod block;

applications/tari_app_grpc/src/conversions/new_block_template.rs

@@ -45,6 +45,7 @@ impl TryFrom<NewBlockTemplate> for grpc::NewBlockTemplate {
                 pow_algo: block.header.pow.pow_algo.as_u64(),
                 pow_data: block.header.pow.pow_data,
             }),
+            validator_node_merkle_root: block.header.validator_node_merkle_root,
         };
         Ok(Self {
             body: Some(grpc::AggregateBody {
@@ -91,6 +92,7 @@ impl TryFrom<grpc::NewBlockTemplate> for NewBlockTemplate {
             total_kernel_offset,
             total_script_offset,
             pow,
+            validator_node_merkle_root: header.validator_node_merkle_root,
         };
         let body = block
             .body

applications/tari_app_grpc/src/conversions/output_features.rs

@@ -22,12 +22,14 @@
 
 use std::convert::{TryFrom, TryInto};
 
+use tari_common_types::types::PublicKey;
 use tari_core::transactions::transaction_components::{
     OutputFeatures,
     OutputFeaturesVersion,
     OutputType,
     SideChainFeatures,
 };
+use tari_utilities::ByteArray;
 
 use crate::tari_rpc as grpc;
 
@@ -46,6 +48,9 @@ impl TryFrom<grpc::OutputFeatures> for OutputFeatures {
             .try_into()
             .map_err(|_| "Invalid output type: overflow")?;
 
+        let validator_node_public_key = PublicKey::from_vec(&features.validator_node_public_key).ok();
+        let validator_node_signature = features.validator_node_signature.map(|s| s.try_into()).transpose()?;
+
         Ok(OutputFeatures::new(
             OutputFeaturesVersion::try_from(
                 u8::try_from(features.version).map_err(|_| "Invalid version: overflowed u8")?,
@@ -54,6 +59,8 @@ impl TryFrom<grpc::OutputFeatures> for OutputFeatures {
             features.maturity,
             features.metadata,
             sidechain_features,
+            validator_node_public_key,
+            validator_node_signature,
         ))
     }
 }
@@ -66,6 +73,14 @@ impl From<OutputFeatures> for grpc::OutputFeatures {
             maturity: features.maturity,
             metadata: features.metadata,
             sidechain_features: features.sidechain_features.map(Into::into),
+            validator_node_public_key: features
+                .validator_node_public_key
+                .map(|pk| pk.as_bytes().to_vec())
+                .unwrap_or_default(),
+            validator_node_signature: features.validator_node_signature.map(|s| grpc::Signature {
+                public_nonce: Vec::from(s.get_public_nonce().as_bytes()),
+                signature: Vec::from(s.get_signature().as_bytes()),
+            }),
         }
     }
 }

applications/tari_base_node/src/builder.rs

@@ -173,9 +173,11 @@ pub async fn configure_and_initialize_node(
 ) -> Result<BaseNodeContext, ExitError> {
     let result = match &app_config.base_node.db_type {
         DatabaseType::Lmdb => {
+            let rules = ConsensusManager::builder(app_config.base_node.network).build();
             let backend = create_lmdb_database(
                 app_config.base_node.lmdb_path.as_path(),
                 app_config.base_node.lmdb.clone(),
+                rules,
             )
             .map_err(|e| ExitError::new(ExitCode::DatabaseError, e))?;
             build_node_context(backend, app_config, node_identity, interrupt_signal).await?

applications/tari_base_node/src/grpc/base_node_grpc_server.rs

@@ -134,6 +134,7 @@ impl BaseNodeGrpcServer {}
 #[tonic::async_trait]
 impl tari_rpc::base_node_server::BaseNode for BaseNodeGrpcServer {
     type FetchMatchingUtxosStream = mpsc::Receiver<Result<tari_rpc::FetchMatchingUtxosResponse, Status>>;
+    type GetActiveValidatorNodesStream = mpsc::Receiver<Result<tari_rpc::ActiveValidatorNode, Status>>;
     type GetBlocksStream = mpsc::Receiver<Result<tari_rpc::HistoricalBlock, Status>>;
     type GetMempoolTransactionsStream = mpsc::Receiver<Result<tari_rpc::GetMempoolTransactionsResponse, Status>>;
     type GetNetworkDifficultyStream = mpsc::Receiver<Result<tari_rpc::NetworkDifficultyResponse, Status>>;
@@ -1576,6 +1577,101 @@ impl tari_rpc::base_node_server::BaseNode for BaseNodeGrpcServer {
 
         Ok(Response::new(response))
     }
+
+    async fn get_committee(
+        &self,
+        request: Request<tari_rpc::GetCommitteeRequest>,
+    ) -> Result<Response<tari_rpc::GetCommitteeResponse>, Status> {
+        let request = request.into_inner();
+        let report_error_flag = self.report_error_flag();
+        debug!(target: LOG_TARGET, "Incoming GRPC request for GetCommittee");
+        let mut handler = self.node_service.clone();
+        let response = handler
+            .get_committee(request.height, request.shard_key.try_into().unwrap())
+            .await
+            .map_err(|e| {
+                error!(target: LOG_TARGET, "Error {}", e);
+                report_error(report_error_flag, Status::internal(e.to_string()))
+            })?
+            .iter()
+            .map(|a| a.shard_key.to_vec())
+            .collect();
+        Ok(Response::new(tari_rpc::GetCommitteeResponse { public_key: response }))
+    }
+
+    async fn get_active_validator_nodes(
+        &self,
+        request: Request<tari_rpc::GetActiveValidatorNodesRequest>,
+    ) -> Result<Response<Self::GetActiveValidatorNodesStream>, Status> {
+        let request = request.into_inner();
+        let report_error_flag = self.report_error_flag();
+        debug!(target: LOG_TARGET, "Incoming GRPC request for GetActiveValidatorNodes");
+
+        let mut handler = self.node_service.clone();
+        let (mut tx, rx) = mpsc::channel(1000);
+
+        task::spawn(async move {
+            let active_validator_nodes = match handler.get_active_validator_nodes(request.height).await {
+                Err(err) => {
+                    warn!(target: LOG_TARGET, "Error communicating with base node: {}", err,);
+                    return;
+                },
+                Ok(data) => data,
+            };
+            for active_validator_node in active_validator_nodes {
+                let active_validator_node = match tari_rpc::ActiveValidatorNode::try_from(active_validator_node) {
+                    Ok(t) => t,
+                    Err(e) => {
+                        warn!(
+                            target: LOG_TARGET,
+                            "Error sending converting active validator node for GRPC: {}", e
+                        );
+                        match tx
+                            .send(Err(report_error(
+                                report_error_flag,
+                                Status::internal("Error converting active validator node"),
+                            )))
+                            .await
+                        {
+                            Ok(_) => (),
+                            Err(send_err) => {
+                                warn!(target: LOG_TARGET, "Error sending error to GRPC client: {}", send_err)
+                            },
+                        }
+                        return;
+                    },
+                };
+
+                match tx.send(Ok(active_validator_node)).await {
+                    Ok(_) => (),
+                    Err(err) => {
+                        warn!(
+                            target: LOG_TARGET,
+                            "Error sending mempool transaction via GRPC:  {}", err
+                        );
+                        match tx
+                            .send(Err(report_error(
+                                report_error_flag,
+                                Status::unknown("Error sending data"),
+                            )))
+                            .await
+                        {
+                            Ok(_) => (),
+                            Err(send_err) => {
+                                warn!(target: LOG_TARGET, "Error sending error to GRPC client: {}", send_err)
+                            },
+                        }
+                        return;
+                    },
+                }
+            }
+        });
+        debug!(
+            target: LOG_TARGET,
+            "Sending GetActiveValidatorNodes response stream to client"
+        );
+        Ok(Response::new(rx))
+    }
 }
 
 enum BlockGroupType {

applications/tari_base_node/src/recovery.rs

@@ -74,22 +74,23 @@ pub fn initiate_recover_db(config: &BaseNodeConfig) -> Result<(), ExitError> {
 
 pub async fn run_recovery(node_config: &BaseNodeConfig) -> Result<(), anyhow::Error> {
     println!("Starting recovery mode");
+    let rules = ConsensusManager::builder(node_config.network).build();
     let (temp_db, main_db, temp_path) = match &node_config.db_type {
         DatabaseType::Lmdb => {
-            let backend = create_lmdb_database(&node_config.lmdb_path, node_config.lmdb.clone()).map_err(|e| {
-                error!(target: LOG_TARGET, "Error opening db: {}", e);
-                anyhow!("Could not open DB: {}", e)
-            })?;
+            let backend = create_lmdb_database(&node_config.lmdb_path, node_config.lmdb.clone(), rules.clone())
+                .map_err(|e| {
+                    error!(target: LOG_TARGET, "Error opening db: {}", e);
+                    anyhow!("Could not open DB: {}", e)
+                })?;
             let temp_path = temp_dir().join("temp_recovery");
 
-            let temp = create_lmdb_database(&temp_path, node_config.lmdb.clone()).map_err(|e| {
+            let temp = create_lmdb_database(&temp_path, node_config.lmdb.clone(), rules.clone()).map_err(|e| {
                 error!(target: LOG_TARGET, "Error opening recovery db: {}", e);
                 anyhow!("Could not open recovery DB: {}", e)
             })?;
             (temp, backend, temp_path)
         },
     };
-    let rules = ConsensusManager::builder(node_config.network).build();
     let factories = CryptoFactories::default();
     let randomx_factory = RandomXFactory::new(node_config.max_randomx_vms);
     let validators = Validators::new(