Additional fixes.

applications/tari_validator_node/Cargo.toml

@@ -32,7 +32,6 @@ blake2 = "0.9.2"
 clap = "2.33.3"
 digest = "0.9.0"
 futures = { version = "^0.3.1" }
-itertools = "0.10.3"
 log = { version = "0.4.8", features = ["std"] }
 lmdb-zero = "0.4.4"
 prost = "0.9"

applications/tari_validator_node/src/asset.rs

@@ -35,7 +35,7 @@ pub struct Asset {
     definition: AssetDefinition,
     current_state: bool,
     // Changes in the committee for this asset.
-    // Mined height of the change TXs, and the involvment in the committe (true = part of committee)
+    // Mined height of the change TXs, and the involvement in the committe (true = part of committee)
     next_states: HashMap<u64, bool>,
     kill_signal: Option<Arc<AtomicBool>>,
 }
@@ -52,14 +52,14 @@ impl Asset {
 
     pub fn update_height<Fstart>(&mut self, height: u64, start: Fstart)
     where Fstart: Fn(AssetDefinition) -> Arc<AtomicBool> {
-        if let Some((&height, &involment)) = self
+        if let Some((&height, &involvement)) = self
             .next_states
             .iter()
             .find(|(&mined_height, _)| mined_height <= height)
         {
             // State change
-            if self.current_state != involment {
-                if involment {
+            if self.current_state != involvement {
+                if involvement {
                     self.kill_signal = Some(start(self.definition.clone()));
                 } else {
                     // Switch on the kill signal for the asset to end processing
@@ -68,7 +68,7 @@ impl Asset {
                     self.kill_signal = None;
                 }
             }
-            self.current_state = involment;
+            self.current_state = involvement;
             // We have the current state set and we will keep only future updates
             self.next_states
                 .retain(|&effective_height, _| effective_height > height);
@@ -81,11 +81,11 @@ impl Asset {
         self.current_state || !self.next_states.is_empty()
     }
 
-    pub fn add_state(&mut self, height: u64, involment: bool) {
-        self.next_states.insert(height, involment);
+    pub fn add_state(&mut self, height: u64, involvement: bool) {
+        self.next_states.insert(height, involvement);
     }
 
-    pub fn get_current_state(&self) -> bool {
+    pub fn is_committee_member(&self) -> bool {
         self.current_state
     }
 }

applications/tari_validator_node/src/dan_node.rs

@@ -25,7 +25,6 @@ use std::{
     time::Duration,
 };
 
-use itertools::Itertools;
 use log::*;
 use tari_common::{
     configuration::ValidatorNodeConfig,
@@ -115,7 +114,7 @@ impl DanNode {
                 } else {
                     next_scanned_height = u64::MAX; // Never run again.
                 }
-                let assets = base_node_client
+                let mut assets = base_node_client
                     .get_assets_for_dan_node(node_identity.public_key().clone())
                     .await
                     .map_err(|e| ExitError::new(ExitCode::DigitalAssetError, e))?;
@@ -138,7 +137,7 @@ impl DanNode {
                     .cloned()
                     .collect::<Vec<PublicKey>>();
                 for public_key in active_public_keys {
-                    if !known_active_public_keys.contains(&public_key) {
+                    if !known_active_public_keys.any(|pk| pk == public_key) {
                         // Active asset is not part of the newly known active assets, maybe there were no checkpoint for
                         // the asset. Are we still part of the committee?
                         if let (false, height) = base_node_client

applications/tari_validator_node/src/monitoring.rs

@@ -50,12 +50,12 @@ impl Monitoring {
         }
     }
 
-    pub fn add_state(&mut self, asset_public_key: PublicKey, height: u64, involment: bool) {
+    pub fn add_state(&mut self, asset_public_key: PublicKey, height: u64, involvement: bool) {
         // Add committee_management_confirmation_time to the mined height = effective height
         self.assets
             .get_mut(&asset_public_key)
             .unwrap()
-            .add_state(height + self.committee_management_confirmation_time, involment);
+            .add_state(height + self.committee_management_confirmation_time, involvement);
     }
 
     pub fn update_height<Fstart: Clone>(&mut self, height: u64, start: Fstart)
@@ -70,7 +70,7 @@ impl Monitoring {
     pub fn get_active_public_keys(&self) -> Vec<&PublicKey> {
         self.assets
             .keys()
-            .filter(|&a| self.assets.get(a).unwrap().get_current_state())
+            .filter(|&a| self.assets.get(a).unwrap().is_committee_member())
             .collect()
     }
 }

dan_layer/core/src/workers/consensus_worker.rs

@@ -134,7 +134,7 @@ impl<TSpecification: ServiceSpecification<Addr = PublicKey>> ConsensusWorker<TSp
             "Consensus worker started for asset '{}'. Tip: {}", self.asset_definition.public_key, self.current_view_id
         );
         let starting_view = self.current_view_id;
-        while !stop.as_ref().load(Ordering::Relaxed) {
+        while !stop.load(Ordering::Relaxed) {
             if let Some(max) = max_views_to_process {
                 if max <= self.current_view_id.0 - starting_view.0 {
                     break;