fixed consensus tests and clippies

applications/tari_indexer/src/event_scanner.rs

@@ -28,7 +28,7 @@ use tari_bor::decode;
 use tari_common::configuration::Network;
 use tari_crypto::tari_utilities::message_format::MessageFormat;
 use tari_dan_app_utilities::consensus_constants::ConsensusConstants;
-use tari_dan_common_types::{committee::Committee, shard::Shard, Epoch, NumPreshards, PeerAddress, ShardGroup};
+use tari_dan_common_types::{committee::Committee, Epoch, NumPreshards, PeerAddress, ShardGroup};
 use tari_dan_p2p::proto::rpc::{GetTransactionResultRequest, PayloadResultStatus, SyncBlocksRequest};
 use tari_dan_storage::consensus_models::{Block, BlockId, Decision, TransactionRecord};
 use tari_engine_types::{

applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs

@@ -39,7 +39,7 @@ use diesel::{
 use diesel_migrations::{EmbeddedMigrations, MigrationHarness};
 use log::*;
 use tari_crypto::tari_utilities::hex::to_hex;
-use tari_dan_common_types::{shard::Shard, substate_type::SubstateType, Epoch, ShardGroup};
+use tari_dan_common_types::{substate_type::SubstateType, Epoch, ShardGroup};
 use tari_dan_storage::{consensus_models::BlockId, StorageError};
 use tari_dan_storage_sqlite::{error::SqliteStorageError, SqliteTransaction};
 use tari_engine_types::substate::SubstateId;

bindings/index.ts

@@ -64,6 +64,7 @@ export * from "./src/types/NonFungibleId";
 export * from "./src/types/NonFungibleIndex";
 export * from "./src/types/NonFungibleIndexAddress";
 export * from "./src/types/NonFungibleToken";
+export * from "./src/types/NumPreshards";
 export * from "./src/types/Ordering";
 export * from "./src/types/OwnerRule";
 export * from "./src/types/PeerAddress";
@@ -81,6 +82,7 @@ export * from "./src/types/RestrictedAccessRule";
 export * from "./src/types/RuleRequirement";
 export * from "./src/types/Shard";
 export * from "./src/types/ShardEvidence";
+export * from "./src/types/ShardGroup";
 export * from "./src/types/Substate";
 export * from "./src/types/SubstateAddress";
 export * from "./src/types/SubstateDestroyed";

bindings/src/types/Block.ts

@@ -4,6 +4,7 @@ import type { Epoch } from "./Epoch";
 import type { NodeHeight } from "./NodeHeight";
 import type { QuorumCertificate } from "./QuorumCertificate";
 import type { Shard } from "./Shard";
+import type { ShardGroup } from "./ShardGroup";
 
 export interface Block {
   id: string;
@@ -12,7 +13,7 @@ export interface Block {
   justify: QuorumCertificate;
   height: NodeHeight;
   epoch: Epoch;
-  shard: Shard;
+  shard_group: ShardGroup;
   proposed_by: string;
   total_leader_fee: number;
   merkle_root: string;

bindings/src/types/CommitteeInfo.ts

@@ -1,7 +1,10 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { NumPreshards } from "./NumPreshards";
+import type { ShardGroup } from "./ShardGroup";
 
 export interface CommitteeInfo {
+  num_shards: NumPreshards;
+  num_shard_group_members: number;
   num_committees: number;
-  num_members: number;
-  shard: number;
+  shard_group: ShardGroup;
 }

bindings/src/types/ForeignProposal.ts

@@ -3,7 +3,7 @@ import type { ForeignProposalState } from "./ForeignProposalState";
 import type { NodeHeight } from "./NodeHeight";
 
 export interface ForeignProposal {
-  shard: number;
+  shard_group: number;
   block_id: string;
   state: ForeignProposalState;
   proposed_height: NodeHeight | null;

bindings/src/types/NumPreshards.ts

@@ -0,0 +1,12 @@
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type NumPreshards =
+  | "One"
+  | "Two"
+  | "Four"
+  | "Eight"
+  | "Sixteen"
+  | "ThirtyTwo"
+  | "SixtyFour"
+  | "OneTwentyEight"
+  | "TwoFiftySix";

bindings/src/types/QuorumCertificate.ts

@@ -2,15 +2,15 @@
 import type { Epoch } from "./Epoch";
 import type { NodeHeight } from "./NodeHeight";
 import type { QuorumDecision } from "./QuorumDecision";
-import type { Shard } from "./Shard";
+import type { ShardGroup } from "./ShardGroup";
 import type { ValidatorSignature } from "./ValidatorSignature";
 
 export interface QuorumCertificate {
   qc_id: string;
   block_id: string;
   block_height: NodeHeight;
   epoch: Epoch;
-  shard: Shard;
+  shard_group: ShardGroup;
   signatures: Array<ValidatorSignature>;
   leaf_hashes: Array<string>;
   decision: QuorumDecision;

bindings/src/types/ShardGroup.ts

@@ -0,0 +1,7 @@
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { Shard } from "./Shard";
+
+export interface ShardGroup {
+  start: Shard;
+  end_inclusive: Shard;
+}

bindings/tsconfig.json

@@ -4,7 +4,7 @@
     "target": "ESNext",
     "moduleResolution": "Bundler",
     "declaration": true,
-    "outDir": "./dist",
+    "outDir": "./dist"
   },
-  "include": ["src/**/*"],
+  "include": ["src/**/*"]
 }

dan_layer/common_types/src/committee.rs

@@ -169,27 +169,34 @@ impl<TAddr: PartialEq> FromIterator<Committee<TAddr>> for Committee<TAddr> {
 )]
 pub struct CommitteeInfo {
     num_shards: NumPreshards,
-    num_members: u32,
+    num_shard_group_members: u32,
+    num_committees: u32,
     shard_group: ShardGroup,
 }
 
 impl CommitteeInfo {
-    pub fn new(num_shards: NumPreshards, num_members: u32, shard_group: ShardGroup) -> Self {
+    pub fn new(
+        num_shards: NumPreshards,
+        num_shard_group_members: u32,
+        num_committees: u32,
+        shard_group: ShardGroup,
+    ) -> Self {
         Self {
             num_shards,
-            num_members,
+            num_shard_group_members,
+            num_committees,
             shard_group,
         }
     }
 
     /// Returns $n - f$ where n is the number of committee members and f is the tolerated failure nodes.
     pub fn quorum_threshold(&self) -> u32 {
-        self.num_members - self.max_failures()
+        self.num_shard_group_members - self.max_failures()
     }
 
     /// Returns the maximum number of failures $f$ that can be tolerated by this committee.
     pub fn max_failures(&self) -> u32 {
-        let len = self.num_members;
+        let len = self.num_shard_group_members;
         if len == 0 {
             return 0;
         }
@@ -200,10 +207,6 @@ impl CommitteeInfo {
         self.num_shards
     }
 
-    pub fn num_members(&self) -> u32 {
-        self.num_members
-    }
-
     pub fn shard_group(&self) -> ShardGroup {
         self.shard_group
     }
@@ -245,11 +248,23 @@ impl CommitteeInfo {
             .filter(|substate_address| self.includes_substate_address(substate_address.borrow()))
     }
 
-    /// Calculates the number of distinct shards for a given shard set
-    pub fn count_distinct_shards<B: Borrow<SubstateAddress>, I: IntoIterator<Item = B>>(&self, shards: I) -> usize {
-        shards
+    /// Calculates the number of distinct shards for the given addresses
+    pub fn count_distinct_shards<B: Borrow<SubstateAddress>, I: IntoIterator<Item = B>>(&self, addresses: I) -> usize {
+        addresses
+            .into_iter()
+            .map(|addr| addr.borrow().to_shard(self.num_shards))
+            .collect::<std::collections::HashSet<_>>()
+            .len()
+    }
+
+    /// Calculates the number of distinct shard groups for the given addresses
+    pub fn count_distinct_shard_groups<B: Borrow<SubstateAddress>, I: IntoIterator<Item = B>>(
+        &self,
+        addresses: I,
+    ) -> usize {
+        addresses
             .into_iter()
-            .map(|shard| shard.borrow().to_shard(self.num_shards))
+            .map(|addr| addr.borrow().to_shard_group(self.num_shards, self.num_committees))
             .collect::<std::collections::HashSet<_>>()
             .len()
     }

dan_layer/consensus/src/hotstuff/block_change_set.rs

@@ -27,7 +27,6 @@ use tari_dan_storage::{
     StorageError,
 };
 use tari_engine_types::substate::SubstateId;
-use tari_state_tree::StateHashTreeDiff;
 use tari_transaction::TransactionId;
 
 #[derive(Debug, Clone)]

dan_layer/consensus/src/hotstuff/common.rs

@@ -19,7 +19,7 @@ use tari_dan_storage::{
     },
     StateStoreReadTransaction,
 };
-use tari_state_tree::{Hash, StateHashTreeDiff, StateTreeError};
+use tari_state_tree::{Hash, StateTreeError};
 
 use crate::{hotstuff::substate_store::ShardedStateTree, traits::LeaderStrategy};
 

dan_layer/consensus/src/hotstuff/config.rs

@@ -2,6 +2,7 @@
 //   SPDX-License-Identifier: BSD-3-Clause
 
 use std::time::Duration;
+
 use tari_common::configuration::Network;
 use tari_dan_common_types::NumPreshards;
 

dan_layer/consensus/src/hotstuff/error.rs

@@ -10,6 +10,7 @@ use tari_dan_storage::{
 use tari_epoch_manager::EpochManagerError;
 use tari_state_tree::StateTreeError;
 use tari_transaction::{TransactionId, VersionedSubstateIdError};
+use tokio::task::JoinError;
 
 use crate::{
     hotstuff::substate_store::SubstateStoreError,
@@ -22,6 +23,8 @@ pub enum HotStuffError {
     StorageError(#[from] StorageError),
     #[error("State tree error: {0}")]
     StateTreeError(#[from] StateTreeError),
+    #[error("Join error: {0}")]
+    JoinError(#[from] JoinError),
     #[error("Internal channel send error when {context}")]
     InternalChannelClosed { context: &'static str },
     #[error("Inbound messaging error: {0}")]

dan_layer/consensus/src/hotstuff/on_propose.rs

@@ -175,7 +175,7 @@ where TConsensusSpec: ConsensusSpec
                 // TODO: This is a hacky workaround, if the executed transaction has no shards after execution, we
                 // remove it from the pool so that it does not get proposed again. Ideally we should be
                 // able to catch this in transaction validation and propose ABORT.
-                if local_committee_info.count_distinct_shards(executed.involved_addresses_iter()) == 0 {
+                if local_committee_info.count_distinct_shard_groups(executed.involved_addresses_iter()) == 0 {
                     self.transaction_pool.remove(tx, *executed.id())?;
                     executed
                         .set_abort("Transaction has no involved shards after execution")
@@ -242,6 +242,7 @@ where TConsensusSpec: ConsensusSpec
     ) -> Result<ExecutedTransaction, HotStuffError> {
         let transaction = TransactionRecord::get(store.read_transaction(), transaction_id)?;
 
+        // TODO: this can fail due to unknown inputs. Need to return an ABORT executed transaction
         let executed = self
             .transaction_executor
             .execute(transaction.into_transaction(), store, current_epoch)
@@ -276,10 +277,10 @@ where TConsensusSpec: ConsensusSpec
             executed_transactions.insert(*executed.id(), executed);
         }
 
-        let num_involved_shards =
-            local_committee_info.count_distinct_shards(tx_rec.evidence().substate_addresses_iter());
+        let num_involved_shard_groups =
+            local_committee_info.count_distinct_shard_groups(tx_rec.evidence().substate_addresses_iter());
 
-        if num_involved_shards == 0 {
+        if num_involved_shard_groups == 0 {
             warn!(
                 target: LOG_TARGET,
                 "Transaction {} has no involved shards, skipping...",
@@ -291,7 +292,7 @@ where TConsensusSpec: ConsensusSpec
 
         // If the transaction is local only, propose LocalOnly. If the transaction is not new, it must have been
         // previously prepared in a multi-shard command (TBD if that a valid thing to do).
-        if num_involved_shards == 1 && !tx_rec.current_stage().is_new() {
+        if num_involved_shard_groups == 1 && !tx_rec.current_stage().is_new() {
             warn!(
                 target: LOG_TARGET,
                 "Transaction {} is local only but was not previously proposed as such. It is in stage {}",
@@ -301,13 +302,13 @@ where TConsensusSpec: ConsensusSpec
         }
 
         // LOCAL-ONLY
-        if num_involved_shards == 1 && tx_rec.current_stage().is_new() {
+        if num_involved_shard_groups == 1 && tx_rec.current_stage().is_new() {
             info!(
                 target: LOG_TARGET,
                 "🏠️ Transaction {} is local only, proposing LocalOnly",
                 tx_rec.transaction_id(),
             );
-            let involved = NonZeroU64::new(num_involved_shards as u64).expect("involved is 1");
+            let involved = NonZeroU64::new(num_involved_shard_groups as u64).expect("involved is 1");
             let leader_fee = tx_rec.calculate_leader_fee(involved, EXHAUST_DIVISOR);
             let tx_atom = tx_rec.get_final_transaction_atom(leader_fee);
             if tx_atom.decision.is_commit() {
@@ -397,7 +398,7 @@ where TConsensusSpec: ConsensusSpec
             // prepared. We can now propose to Accept it. We also propose the decision change which everyone
             // should agree with if they received the same foreign LocalPrepare.
             TransactionPoolStage::LocalPrepared => {
-                let involved = NonZeroU64::new(num_involved_shards as u64).ok_or_else(|| {
+                let involved = NonZeroU64::new(num_involved_shard_groups as u64).ok_or_else(|| {
                     HotStuffError::InvariantError(format!(
                         "Number of involved shards is zero for transaction {}",
                         tx_rec.transaction_id(),

dan_layer/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs

@@ -13,7 +13,6 @@ use tari_dan_storage::{
         BlockId,
         Command,
         Decision,
-        ForeignProposal,
         LastExecuted,
         LastVoted,
         LockedBlock,
@@ -211,7 +210,7 @@ where TConsensusSpec: ConsensusSpec
 
         for cmd in block.commands() {
             if let Some(foreign_proposal) = cmd.foreign_proposal() {
-                if !ForeignProposal::exists(tx, foreign_proposal)? {
+                if !foreign_proposal.exists(tx)? {
                     warn!(
                         target: LOG_TARGET,
                         "❌ Foreign proposal for block {block_id} from bucket {bucket} does not exist in the store",
@@ -632,9 +631,9 @@ where TConsensusSpec: ConsensusSpec
                         return Ok(proposed_block_change_set.no_vote());
                     }
 
-                    let distinct_shards =
-                        local_committee_info.count_distinct_shards(tx_rec.evidence().substate_addresses_iter());
-                    let distinct_shards = NonZeroU64::new(distinct_shards as u64).ok_or_else(|| {
+                    let distinct_shard_groups =
+                        local_committee_info.count_distinct_shard_groups(tx_rec.evidence().substate_addresses_iter());
+                    let distinct_shards = NonZeroU64::new(distinct_shard_groups as u64).ok_or_else(|| {
                         HotStuffError::InvariantError(format!(
                             "Distinct shards is zero for transaction {} in block {}",
                             tx_rec.transaction_id(),