fixes for global transaction, added new consensus test

applications/tari_dan_app_utilities/src/base_layer_scanner.rs

@@ -475,6 +475,8 @@ impl<TAddr: NodeAddressable + 'static> BaseLayerScanner<TAddr> {
             target: LOG_TARGET,
             "🌠 new template found with address {} at height {}", template_address, block_info.height
         );
+        let consensus_constants = self.epoch_manager.get_base_layer_consensus_constants().await?;
+        let epoch = consensus_constants.height_to_epoch(block_info.height);
         self.template_manager
             .add_template(
                 registration.author_public_key,
@@ -484,6 +486,7 @@ impl<TAddr: NodeAddressable + 'static> BaseLayerScanner<TAddr> {
                     registration.binary_sha,
                 ),
                 Some(template_name),
+                epoch,
             )
             .await?;
 

applications/tari_dan_app_utilities/src/template_manager/implementation/manager.rs

@@ -26,7 +26,12 @@ use chrono::Utc;
 use log::*;
 use tari_common_types::types::{FixedHash, PublicKey};
 use tari_crypto::tari_utilities::ByteArray;
-use tari_dan_common_types::{optional::Optional, services::template_provider::TemplateProvider, NodeAddressable};
+use tari_dan_common_types::{
+    optional::Optional,
+    services::template_provider::TemplateProvider,
+    Epoch,
+    NodeAddressable,
+};
 use tari_dan_engine::{
     flow::FlowFactory,
     function_definitions::FlowFunctionDefinition,
@@ -192,6 +197,7 @@ impl<TAddr: NodeAddressable> TemplateManager<TAddr> {
         template: TemplateExecutable,
         template_name: Option<String>,
         template_status: Option<TemplateStatus>,
+        epoch: Epoch,
     ) -> Result<(), TemplateManagerError> {
         enum TemplateHash {
             Hash(Hash),
@@ -244,6 +250,7 @@ impl<TAddr: NodeAddressable> TemplateManager<TAddr> {
             flow_json,
             manifest,
             url: template_url,
+            epoch,
         };
 
         let mut tx = self.global_db.create_transaction()?;
@@ -326,21 +333,6 @@ impl<TAddr: NodeAddressable + Send + Sync + 'static> TemplateProvider for Templa
 
         Ok(Some(loaded))
     }
-
-    fn add_wasm_template(
-        &self,
-        author_public_key: PublicKey,
-        template_address: tari_engine_types::TemplateAddress,
-        template: &[u8],
-    ) -> Result<(), Self::Error> {
-        self.add_template(
-            author_public_key,
-            template_address,
-            TemplateExecutable::CompiledWasm(template.to_vec()),
-            None,
-            Some(TemplateStatus::Active),
-        )
-    }
 }
 
 impl<TAddr> Clone for TemplateManager<TAddr> {

applications/tari_dan_app_utilities/src/template_manager/implementation/service.rs

@@ -25,7 +25,7 @@
 
 use log::*;
 use tari_common_types::types::PublicKey;
-use tari_dan_common_types::{services::template_provider::TemplateProvider, NodeAddressable};
+use tari_dan_common_types::{services::template_provider::TemplateProvider, Epoch, NodeAddressable};
 use tari_dan_engine::function_definitions::FlowFunctionDefinition;
 use tari_dan_storage::global::{DbTemplateType, DbTemplateUpdate, TemplateStatus};
 use tari_engine_types::calculate_template_binary_hash;
@@ -124,11 +124,12 @@ impl<TAddr: NodeAddressable + 'static> TemplateManagerService<TAddr> {
                 template_address,
                 template,
                 template_name,
+                epoch,
                 reply,
             } => {
                 handle(
                     reply,
-                    self.handle_add_template(author_public_key, template_address, template, template_name)
+                    self.handle_add_template(author_public_key, template_address, template, template_name, epoch)
                         .await,
                 );
             },
@@ -243,6 +244,7 @@ impl<TAddr: NodeAddressable + 'static> TemplateManagerService<TAddr> {
         template_address: tari_engine_types::TemplateAddress,
         template: TemplateExecutable,
         template_name: Option<String>,
+        epoch: Epoch,
     ) -> Result<(), TemplateManagerError> {
         let template_status = if matches!(template, TemplateExecutable::DownloadableWasm(_, _)) {
             TemplateStatus::New
@@ -255,6 +257,7 @@ impl<TAddr: NodeAddressable + 'static> TemplateManagerService<TAddr> {
             template.clone(),
             template_name,
             Some(template_status),
+            epoch,
         )?;
 
         // TODO: remove when we remove support for base layer template registration

applications/tari_dan_app_utilities/src/template_manager/interface/handle.rs

@@ -21,6 +21,7 @@
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use tari_common_types::types::PublicKey;
+use tari_dan_common_types::Epoch;
 use tari_template_lib::models::TemplateAddress;
 use tari_validator_node_client::types::TemplateAbi;
 use tokio::sync::{mpsc, oneshot};
@@ -70,6 +71,7 @@ impl TemplateManagerHandle {
         template_address: TemplateAddress,
         template: TemplateExecutable,
         template_name: Option<String>,
+        epoch: Epoch,
     ) -> Result<(), TemplateManagerError> {
         let (tx, rx) = oneshot::channel();
         self.request_tx
@@ -78,6 +80,7 @@ impl TemplateManagerHandle {
                 template_address,
                 template,
                 template_name,
+                epoch,
                 reply: tx,
             })
             .await

applications/tari_dan_app_utilities/src/template_manager/interface/types.rs

@@ -22,6 +22,7 @@
 
 use reqwest::Url;
 use tari_common_types::types::{FixedHash, PublicKey};
+use tari_dan_common_types::Epoch;
 use tari_dan_storage::global::{DbTemplate, DbTemplateType};
 use tari_template_lib::models::TemplateAddress;
 use tari_validator_node_client::types::TemplateAbi;
@@ -91,6 +92,7 @@ pub enum TemplateManagerRequest {
         template_address: tari_engine_types::TemplateAddress,
         template: TemplateExecutable,
         template_name: Option<String>,
+        epoch: Epoch,
         reply: oneshot::Sender<Result<(), TemplateManagerError>>,
     },
     GetTemplate {

applications/tari_validator_node/src/bootstrap.rs

@@ -336,7 +336,6 @@ pub async fn spawn_services(
     handles.push(consensus_join_handle);
 
     let (mempool, join_handle) = mempool::spawn(
-        consensus_constants.num_preshards,
         epoch_manager.clone(),
         create_mempool_transaction_validator(template_manager.clone()),
         state_store.clone(),

applications/tari_validator_node/src/dan_node.rs

@@ -94,7 +94,6 @@ impl DanNode {
             .services
             .state_store
             .with_read_tx(|tx| block.get_committing_transactions(tx))?;
-
         let templates = transactions
             .into_iter()
             .filter_map(|record| {
@@ -115,6 +114,7 @@ impl DanNode {
                 })
             });
 
+        let epoch = self.services.consensus_handle.current_epoch();
         // adding templates to template manager
         let mut template_counter = 0;
         for (author_pub_key, template_address, template) in templates {
@@ -127,7 +127,7 @@ impl DanNode {
             if let Err(err) = self
                 .services
                 .template_manager
-                .add_template(author_pub_key, template_address, template, None)
+                .add_template(author_pub_key, template_address, template, None, epoch)
                 .await
             {
                 error!(target: LOG_TARGET, "🚨Failed to add template: {}", err);

applications/tari_validator_node/src/p2p/services/mempool/gossip.rs

@@ -5,7 +5,7 @@ use std::{collections::HashSet, iter};
 
 use libp2p::{gossipsub, PeerId};
 use log::*;
-use tari_dan_common_types::{Epoch, NumPreshards, PeerAddress, ShardGroup, ToSubstateAddress};
+use tari_dan_common_types::{Epoch, PeerAddress, ShardGroup, ToSubstateAddress};
 use tari_dan_p2p::{proto, DanMessage, NewTransactionMessage, TariMessagingSpec};
 use tari_epoch_manager::{base_layer::EpochManagerHandle, EpochManagerReader};
 use tari_networking::{NetworkingHandle, NetworkingService};
@@ -47,7 +47,6 @@ impl MempoolGossipCodec {
 
 #[derive(Debug)]
 pub(super) struct MempoolGossip<TAddr> {
-    num_preshards: NumPreshards,
     epoch_manager: EpochManagerHandle<TAddr>,
     is_subscribed: Option<ShardGroup>,
     networking: NetworkingHandle<TariMessagingSpec>,
@@ -57,13 +56,11 @@ pub(super) struct MempoolGossip<TAddr> {
 
 impl MempoolGossip<PeerAddress> {
     pub fn new(
-        num_preshards: NumPreshards,
         epoch_manager: EpochManagerHandle<PeerAddress>,
         networking: NetworkingHandle<TariMessagingSpec>,
         rx_gossip: mpsc::UnboundedReceiver<(PeerId, gossipsub::Message)>,
     ) -> Self {
         Self {
-            num_preshards,
             epoch_manager,
             is_subscribed: None,
             networking,
@@ -143,28 +140,37 @@ impl MempoolGossip<PeerAddress> {
         exclude_shard_group: Option<ShardGroup>,
     ) -> Result<(), MempoolError> {
         let n = self.epoch_manager.get_num_committees(epoch).await?;
-        let committee_shard = self.epoch_manager.get_local_committee_info(epoch).await?;
-        let local_shard_group = committee_shard.shard_group();
-        let shard_groups = msg
-            .transaction
-            .all_inputs_iter()
-            .map(|s| {
-                s.or_zero_version()
-                    .to_substate_address()
-                    .to_shard_group(self.num_preshards, n)
-            })
-            .chain(iter::once(
-                msg.transaction
-                    .id()
-                    .to_substate_address()
-                    .to_shard_group(self.num_preshards, n),
-            ))
-            .filter(|sg| exclude_shard_group.as_ref() != Some(sg) && sg != &local_shard_group)
-            .collect::<HashSet<_>>();
-        // If the only shard group involved is the excluded one.
-        if shard_groups.is_empty() {
-            return Ok(());
-        }
+        let committee_info = self.epoch_manager.get_local_committee_info(epoch).await?;
+        let local_shard_group = committee_info.shard_group();
+        let shard_groups = if msg.transaction.is_global() {
+            Box::new(
+                committee_info
+                    .all_shard_groups_iter()
+                    .filter(|sg| exclude_shard_group.as_ref() != Some(sg) && sg != &local_shard_group),
+            )
+        } else {
+            let shard_groups = msg
+                .transaction
+                .all_inputs_iter()
+                .map(|s| {
+                    s.or_zero_version()
+                        .to_substate_address()
+                        .to_shard_group(committee_info.num_preshards(), n)
+                })
+                .chain(iter::once(
+                    msg.transaction
+                        .id()
+                        .to_substate_address()
+                        .to_shard_group(committee_info.num_preshards(), n),
+                ))
+                .filter(|sg| exclude_shard_group.as_ref() != Some(sg) && sg != &local_shard_group)
+                .collect::<HashSet<_>>();
+            // If the only shard group involved is the excluded one.
+            if shard_groups.is_empty() {
+                return Ok(());
+            }
+            Box::new(shard_groups.into_iter()) as Box<dyn Iterator<Item = ShardGroup> + Send>
+        };
 
         let msg = self
             .codec

applications/tari_validator_node/src/p2p/services/mempool/initializer.rs

@@ -22,7 +22,7 @@
 
 use libp2p::{gossipsub, PeerId};
 use log::*;
-use tari_dan_common_types::{NumPreshards, PeerAddress};
+use tari_dan_common_types::PeerAddress;
 use tari_dan_p2p::TariMessagingSpec;
 use tari_epoch_manager::base_layer::EpochManagerHandle;
 use tari_networking::NetworkingHandle;
@@ -42,7 +42,6 @@ use crate::{
 const LOG_TARGET: &str = "tari::dan::validator_node::mempool";
 
 pub fn spawn<TValidator>(
-    num_preshards: NumPreshards,
     epoch_manager: EpochManagerHandle<PeerAddress>,
     transaction_validator: TValidator,
     state_store: SqliteStateStore<PeerAddress>,
@@ -61,7 +60,6 @@ where
     #[cfg(feature = "metrics")]
     let metrics = PrometheusMempoolMetrics::new(metrics_registry);
     let mempool = MempoolService::new(
-        num_preshards,
         rx_mempool_request,
         epoch_manager,
         transaction_validator,

applications/tari_validator_node/src/p2p/services/mempool/service.rs

@@ -25,7 +25,7 @@ use std::{collections::HashSet, fmt::Display, iter};
 use libp2p::{gossipsub, PeerId};
 use log::*;
 use tari_consensus::hotstuff::HotstuffEvent;
-use tari_dan_common_types::{optional::Optional, NumPreshards, PeerAddress, ShardGroup, ToSubstateAddress};
+use tari_dan_common_types::{optional::Optional, PeerAddress, ShardGroup, ToSubstateAddress};
 use tari_dan_p2p::{DanMessage, NewTransactionMessage, TariMessagingSpec};
 use tari_dan_storage::{consensus_models::TransactionRecord, StateStore};
 use tari_engine_types::commit_result::RejectReason;
@@ -67,7 +67,6 @@ impl<TValidator> MempoolService<TValidator>
 where TValidator: Validator<Transaction, Context = (), Error = TransactionValidationError>
 {
     pub(super) fn new(
-        num_preshards: NumPreshards,
         mempool_requests: mpsc::Receiver<MempoolRequest>,
         epoch_manager: EpochManagerHandle<PeerAddress>,
         before_execute_validator: TValidator,
@@ -78,7 +77,7 @@ where TValidator: Validator<Transaction, Context = (), Error = TransactionValida
         #[cfg(feature = "metrics")] metrics: PrometheusMempoolMetrics,
     ) -> Self {
         Self {
-            gossip: MempoolGossip::new(num_preshards, epoch_manager.clone(), networking, rx_gossip),
+            gossip: MempoolGossip::new(epoch_manager.clone(), networking, rx_gossip),
             transactions: Default::default(),
             mempool_requests,
             epoch_manager,

dan_layer/common_types/src/committee.rs

@@ -232,6 +232,9 @@ impl CommitteeInfo {
     }
 
     pub fn includes_substate_id(&self, substate_id: &SubstateId) -> bool {
+        if substate_id.is_global() {
+            return true;
+        }
         // version doesnt affect shard
         let addr = SubstateAddress::from_substate_id(substate_id, 0);
         let shard = addr.to_shard(self.num_shards);
@@ -265,6 +268,10 @@ impl CommitteeInfo {
             .into_iter()
             .filter(|substate_address| self.includes_substate_address(substate_address.borrow()))
     }
+
+    pub fn all_shard_groups_iter(&self) -> impl Iterator<Item = ShardGroup> {
+        self.num_shards.all_shard_groups_iter(self.num_committees)
+    }
 }
 
 #[derive(Debug, Clone, Serialize)]

dan_layer/common_types/src/lock_intent.rs

@@ -51,6 +51,10 @@ impl SubstateLockType {
     pub fn is_output(&self) -> bool {
         matches!(self, Self::Output)
     }
+
+    pub fn is_input(&self) -> bool {
+        !self.is_output()
+    }
 }
 
 impl fmt::Display for SubstateLockType {