feat: add chain storage and mempool validation for unique asset ids (#…

…3416) * test: fix test compilation * test: fix failing tests * test: ignore 2 failing tests that need to be updated with new db structure * feat: add unique id index to lmdb cargo fmt Rebase fixes * ignore some failing tests to be fixed * feat: add chain storage and mempool validation for unique asset ids * chore: fix compile issue after merge * fmt * fix * compile fixes * doctest * clippy --fix Co-authored-by: David Main <51991544+StriderDM@users.noreply.github.com>
tari-project · Oct 5, 2021 · 7331b0c · 7331b0c
1 parent 8082000
commit 7331b0c
Show file tree

Hide file tree

Showing 79 changed files with 909 additions and 506 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/applications/tari_app_grpc/src/conversions/output_features.rs b/applications/tari_app_grpc/src/conversions/output_features.rs
@@ -60,10 +60,7 @@ impl From<OutputFeatures> for grpc::OutputFeatures {
             maturity: features.maturity,
             metadata: features.metadata,
             unique_id: features.unique_id,
-            parent_public_key: match features.parent_public_key {
-                Some(a) => Some(a.as_bytes().to_vec()),
-                None => None,
-            },
+            parent_public_key: features.parent_public_key.map(|a| a.as_bytes().to_vec()),
             asset: features.asset.map(|a| a.into()),
             mint_non_fungible: features.mint_non_fungible.map(|m| m.into()),
             sidechain_checkpoint: features.sidechain_checkpoint.map(|m| m.into()),

diff --git a/applications/tari_base_node/src/grpc/base_node_grpc_server.rs b/applications/tari_base_node/src/grpc/base_node_grpc_server.rs
@@ -604,6 +604,7 @@ impl tari_rpc::base_node_server::BaseNode for BaseNodeGrpcServer {
             },
             TxStorageResponse::NotStored |
             TxStorageResponse::NotStoredOrphan |
+            TxStorageResponse::NotStoredConsensus |
             TxStorageResponse::NotStoredTimeLocked => tari_rpc::SubmitTransactionResponse {
                 result: tari_rpc::SubmitTransactionResult::Rejected.into(),
             },
@@ -669,6 +670,7 @@ impl tari_rpc::base_node_server::BaseNode for BaseNodeGrpcServer {
                 }
             },
             TxStorageResponse::NotStored |
+            TxStorageResponse::NotStoredConsensus |
             TxStorageResponse::NotStoredOrphan |
             TxStorageResponse::NotStoredTimeLocked => tari_rpc::TransactionStateResponse {
                 result: tari_rpc::TransactionLocation::NotStored.into(),

diff --git a/applications/tari_console_wallet/src/ui/components/send_tab.rs b/applications/tari_console_wallet/src/ui/components/send_tab.rs
@@ -133,7 +133,7 @@ impl SendTab {
 
         let amount_input = Paragraph::new(match &self.selected_unique_id {
             Some(token) => format!("Token selected : {}", token.to_hex()),
-            None => format!("{}", self.amount_field),
+            None => self.amount_field.to_string(),
         })
         .style(match self.send_input_mode {
             SendInputMode::Amount => Style::default().fg(Color::Magenta),
@@ -800,7 +800,7 @@ impl<B: Backend> Component<B> for SendTab {
             } else {
                 let tokens: Vec<&Token> = app_state
                     .get_owned_tokens()
-                    .into_iter()
+                    .iter()
                     .filter(|&token| token.output_status() == "Unspent")
                     .collect();
                 self.selected_unique_id = Some(Vec::from(tokens[index - 1].unique_id()));
@@ -817,7 +817,7 @@ impl<B: Backend> Component<B> for SendTab {
             let index = self.table_state.selected().map(|s| s + 1).unwrap_or_default();
             let tokens: Vec<&Token> = app_state
                 .get_owned_tokens()
-                .into_iter()
+                .iter()
                 .filter(|&token| token.output_status() == "Unspent")
                 .collect();
             if index > tokens.len().saturating_sub(1) {

diff --git a/applications/tari_console_wallet/src/ui/components/transactions_tab.rs b/applications/tari_console_wallet/src/ui/components/transactions_tab.rs
@@ -359,7 +359,7 @@ impl TransactionsTab {
                 };
             let direction = Span::styled(format!("{}", tx.direction), Style::default().fg(Color::White));
             let amount = Span::styled(
-                format!("{}", match tx.get_unique_id() {
+                (match tx.get_unique_id() {
                     Some(unique_id) => unique_id,
                     None => tx.amount.to_string(),
                 }),

diff --git a/applications/tari_validator_node/Cargo.toml b/applications/tari_validator_node/Cargo.toml
@@ -17,7 +17,7 @@ tari_mmr = { path = "../../base_layer/mmr" }
 tari_p2p = { path = "../../base_layer/p2p" }
 tari_service_framework = { path = "../../base_layer/service_framework" }
 tari_shutdown = { path = "../../infrastructure/shutdown" }
-tari_storage = { version = "^0.9", path = "../../infrastructure/storage" }
+tari_storage = { path = "../../infrastructure/storage" }
 
 anyhow = "1.0.32"
 async-trait = "0.1.50"
@@ -30,7 +30,8 @@ prost = "0.8"
 prost-types = "0.8"
 serde = "1.0.126"
 thiserror = "^1.0.20"
-tokio = { version = "0.2.10", features = ["macros", "sync", "time"] }
+tokio = { version="1.10", features = ["macros", "time"]}
+tokio-stream = { version = "0.1.7", features = ["sync"] }
 tonic = "0.5.2"
 
 # saving of patricia tree

diff --git a/applications/tari_validator_node/src/cmd_args.rs b/applications/tari_validator_node/src/cmd_args.rs
@@ -23,7 +23,7 @@
 use clap::App;
 
 pub fn get_operation_mode() -> OperationMode {
-    let matches = App::new("Tari DAN node").version("1.0").get_matches();
+    let _matches = App::new("Tari DAN node").version("1.0").get_matches();
     OperationMode::Run
 }
 

diff --git a/applications/tari_validator_node/src/dan_layer/dan_node.rs b/applications/tari_validator_node/src/dan_layer/dan_node.rs
@@ -37,7 +37,6 @@ use crate::{
         storage::{AssetDataStore, LmdbAssetStore},
         workers::ConsensusWorker,
     },
-    digital_assets_error::DigitalAssetError,
     ExitCodes,
 };
 use log::*;
@@ -47,7 +46,7 @@ use tari_app_utilities::{
     identity_management::{load_from_json, setup_node_identity},
     utilities::convert_socks_authentication,
 };
-use tari_common::{CommsTransport, ConfigBootstrap, GlobalConfig, TorControlAuthentication};
+use tari_common::{CommsTransport, GlobalConfig, TorControlAuthentication};
 use tari_comms::{
     peer_manager::PeerFeatures,
     socks,
@@ -60,16 +59,16 @@ use tari_comms::{
     UnspawnedCommsNode,
 };
 use tari_comms_dht::{DbConnectionUrl, Dht, DhtConfig};
-use tari_crypto::tari_utilities::hex::{Hex, HexError};
+use tari_crypto::tari_utilities::hex::{Hex};
 use tari_p2p::{
-    comms_connector::{pubsub_connector, PubsubDomainConnector, SubscriptionFactory},
+    comms_connector::{pubsub_connector, SubscriptionFactory},
     initialization::{spawn_comms_using_transport, P2pConfig, P2pInitializer},
     tari_message::TariMessageType,
     transport::{TorConfig, TransportType},
 };
 use tari_service_framework::{ServiceHandles, StackBuilder};
-use tari_shutdown::{Shutdown, ShutdownSignal};
-use tokio::{runtime::Handle, task};
+use tari_shutdown::{ShutdownSignal};
+use tokio::{task};
 
 const LOG_TARGET: &str = "tari::dan::dan_node";
 
@@ -199,7 +198,7 @@ impl DanNode {
     fn create_comms_config(&self, node_identity: Arc<NodeIdentity>) -> P2pConfig {
         P2pConfig {
             network: self.config.network,
-            node_identity: node_identity.clone(),
+            node_identity,
             transport_type: self.create_transport_type(),
             datastore_path: self.config.peer_db_path.clone(),
             peer_database_name: "peers".to_string(),

diff --git a/applications/tari_validator_node/src/dan_layer/models/hot_stuff_message.rs b/applications/tari_validator_node/src/dan_layer/models/hot_stuff_message.rs
@@ -22,7 +22,7 @@
 
 use crate::dan_layer::models::{HotStuffMessageType, HotStuffTreeNode, Payload, QuorumCertificate, Signature, ViewId};
 use digest::Digest;
-use std::hash::Hash;
+
 use tari_crypto::common::Blake256;
 
 #[derive(Debug, Clone)]

diff --git a/applications/tari_validator_node/src/dan_layer/models/hot_stuff_tree_node.rs b/applications/tari_validator_node/src/dan_layer/models/hot_stuff_tree_node.rs
@@ -23,7 +23,6 @@
 use crate::dan_layer::models::{Payload, TreeNodeHash};
 use digest::Digest;
 use std::{
-    collections::hash_map::DefaultHasher,
     hash::{Hash, Hasher},
 };
 use tari_crypto::common::Blake256;

diff --git a/applications/tari_validator_node/src/dan_layer/models/instruction.rs b/applications/tari_validator_node/src/dan_layer/models/instruction.rs
@@ -46,7 +46,7 @@ impl PartialEq for Instruction {
 }
 
 impl Instruction {
-    pub fn new(asset_id: PublicKey, method: String, args: Vec<Vec<u8>>, from: TokenId, signature: ComSig) -> Self {
+    pub fn new(asset_id: PublicKey, method: String, args: Vec<Vec<u8>>, from: TokenId, _signature: ComSig) -> Self {
         let mut s = Self {
             asset_id,
             method,

diff --git a/applications/tari_validator_node/src/dan_layer/models/instruction_set.rs b/applications/tari_validator_node/src/dan_layer/models/instruction_set.rs
@@ -44,7 +44,7 @@ pub struct InstructionSet {
 
 impl InstructionSet {
     pub fn empty() -> Self {
-        Self::from_slice(&vec![])
+        Self::from_slice(&[])
     }
 
     pub fn from_slice(instructions: &[Instruction]) -> Self {

diff --git a/applications/tari_validator_node/src/dan_layer/models/mod.rs b/applications/tari_validator_node/src/dan_layer/models/mod.rs
@@ -20,7 +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.
 
-use std::cmp::Ordering;
+
 
 mod block;
 mod committee;
@@ -44,8 +44,7 @@ pub use quorum_certificate::QuorumCertificate;
 pub use replica_info::ReplicaInfo;
 use std::{
     convert::TryFrom,
-    fmt,
-    fmt::{Debug, Formatter},
+    fmt::{Debug},
     hash::Hash,
 };
 pub use view::View;

diff --git a/...lidator_node/src/dan_layer/services/infrastructure_services/inbound_connection_service.rs b/...lidator_node/src/dan_layer/services/infrastructure_services/inbound_connection_service.rs
@@ -20,7 +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.
 
-use crate::dan_layer::models::{HotStuffMessage, InstructionSet, Payload, ViewId};
+use crate::dan_layer::models::{HotStuffMessage, InstructionSet, Payload};
 
 use crate::{
     dan_layer::services::infrastructure_services::NodeAddressable,
@@ -29,9 +29,9 @@ use crate::{
 };
 use async_trait::async_trait;
 use futures::{self, pin_mut, Stream, StreamExt};
-use std::{convert::TryInto, marker::PhantomData, sync::Arc};
+use std::{convert::TryInto, sync::Arc};
 use tari_comms::types::CommsPublicKey;
-use tari_p2p::{comms_connector::PeerMessage, domain_message::DomainMessage};
+use tari_p2p::{comms_connector::PeerMessage};
 use tari_shutdown::ShutdownSignal;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 
@@ -61,16 +61,12 @@ impl TariCommsInboundConnectionService {
 
     pub fn take_receiver(&mut self) -> Option<TariCommsInboundReceiver<InstructionSet>> {
         // Takes the receiver, can only be done once
-        if let Some(receiver) = self.receiver.take() {
-            Some(receiver)
-        } else {
-            None
-        }
+        self.receiver.take()
     }
 
     pub async fn run(
         &mut self,
-        shutdown_signal: ShutdownSignal,
+        _shutdown_signal: ShutdownSignal,
         inbound_stream: impl Stream<Item = Arc<PeerMessage>>,
     ) -> Result<(), DigitalAssetError> {
         let inbound_stream = inbound_stream.fuse();
@@ -98,7 +94,7 @@ impl TariCommsInboundConnectionService {
         let proto_message: dan_p2p::HotStuffMessage = message.decode_message().unwrap();
         let hot_stuff_message = proto_message
             .try_into()
-            .map_err(|s| DigitalAssetError::InvalidPeerMessage(s))?;
+            .map_err(DigitalAssetError::InvalidPeerMessage)?;
         self.sender.send((from, hot_stuff_message)).await.unwrap();
         Ok(())
     }

diff --git a/applications/tari_validator_node/src/dan_layer/services/infrastructure_services/mocks/mod.rs b/applications/tari_validator_node/src/dan_layer/services/infrastructure_services/mocks/mod.rs
@@ -22,13 +22,13 @@
 
 use crate::{
     dan_layer::{
-        models::{Committee, HotStuffMessage},
+        models::{HotStuffMessage},
         services::infrastructure_services::{InboundConnectionService, NodeAddressable, OutboundService},
     },
     digital_assets_error::DigitalAssetError,
 };
 use async_trait::async_trait;
-use std::collections::{HashMap, VecDeque};
+use std::collections::{HashMap};
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 
 pub fn mock_inbound<TAddr: NodeAddressable, TPayload: Payload>() -> MockInboundConnectionService<TAddr, TPayload> {
@@ -106,7 +106,7 @@ impl<TAddr: NodeAddressable, TPayload: Payload> MockOutboundService<TAddr, TPayl
 }
 
 use crate::dan_layer::models::Payload;
-use std::{fmt::Debug, hash::Hash};
+use std::{fmt::Debug};
 
 #[async_trait]
 impl<TAddr: NodeAddressable + Send + Sync + Debug, TPayload: Payload> OutboundService<TAddr, TPayload>
@@ -136,7 +136,7 @@ impl<TAddr: NodeAddressable + Send + Sync + Debug, TPayload: Payload> OutboundSe
         _committee: &[TAddr],
         message: HotStuffMessage<TPayload>,
     ) -> Result<(), DigitalAssetError> {
-        let receivers: Vec<TAddr> = self.inbound_senders.keys().map(|k| k.clone()).collect();
+        let receivers: Vec<TAddr> = self.inbound_senders.keys().cloned().collect();
         for receiver in receivers {
             self.send(from.clone(), receiver.clone(), message.clone()).await?
         }

diff --git a/applications/tari_validator_node/src/dan_layer/services/infrastructure_services/mod.rs b/applications/tari_validator_node/src/dan_layer/services/infrastructure_services/mod.rs
@@ -27,7 +27,7 @@ mod outbound_service;
 pub use inbound_connection_service::{InboundConnectionService, TariCommsInboundConnectionService};
 pub use node_addressable::NodeAddressable;
 pub use outbound_service::{OutboundService, TariCommsOutboundService};
-use std::{fmt::Debug, hash::Hash};
+
 
 #[cfg(test)]
 pub mod mocks;
diff --git a/...ns/tari_validator_node/src/dan_layer/services/infrastructure_services/outbound_service.rs b/...ns/tari_validator_node/src/dan_layer/services/infrastructure_services/outbound_service.rs
@@ -23,13 +23,13 @@
 use crate::{
     dan_layer::{
         models::{HotStuffMessage, InstructionSet, Payload},
-        services::infrastructure_services::{InboundConnectionService, NodeAddressable},
+        services::infrastructure_services::{NodeAddressable},
     },
     digital_assets_error::DigitalAssetError,
     p2p,
 };
 use async_trait::async_trait;
-use futures::{future::try_join_all, stream::FuturesUnordered};
+
 use std::marker::PhantomData;
 use tari_comms::types::CommsPublicKey;
 use tari_comms_dht::{domain_message::OutboundDomainMessage, outbound::OutboundMessageRequester};