fix(consensus)!: substate pledges (#1119)

Description
---
fix(consensus)!: multi shard substate pledges
Version is not considered when calculating which preshard a substate
belongs to
Fix several race conditions in consensus
Adds a test for all local inputs, foreign outputs
Build transactions with inputs in tests
Fixed mempool gossip

Motivation and Context
---

Implements cross-shard exchange and pledges for inputs/outputs.

TODO: 
- output shard group does not sequence an unknown transaction from
foreign proposals (PR #1120)
- number of cucumbers scenarios do not include any inputs. This is now
completely invalid, so they have been ignored for now until we switch
them to use the wallet daemon.

How Has This Been Tested?
---
Manually - multi-shard-group test, existing tests, new tests

What process can a PR reviewer use to test or verify this change?
---
Submit transactions to a multi-shard-group network

Breaking Changes
---

- [ ] None
- [x] Requires data directory to be deleted
- [x] Other - Please specify

BREAKING CHANGE: SubstateAddress has increased in size by 4 bytes, new
commands

applications/tari_dan_app_utilities/src/transaction_executor.rs

@@ -1,10 +1,7 @@
 //    Copyright 2023 The Tari Project
 //    SPDX-License-Identifier: BSD-3-Clause
 
-use std::{
-    sync::Arc,
-    time::{Duration, Instant},
-};
+use std::sync::Arc;
 
 use indexmap::IndexMap;
 use log::*;
@@ -19,10 +16,10 @@ use tari_dan_engine::{
     template::LoadedTemplate,
     transaction::{TransactionError, TransactionProcessor},
 };
-use tari_dan_storage::consensus_models::{SubstateLockFlag, VersionedSubstateIdLockIntent};
+use tari_dan_storage::consensus_models::{SubstateLockType, VersionedSubstateIdLockIntent};
 use tari_engine_types::{
-    commit_result::{ExecuteResult, FinalizeResult, RejectReason},
-    substate::Substate,
+    commit_result::ExecuteResult,
+    substate::{Substate, SubstateId},
     virtual_substate::VirtualSubstates,
 };
 use tari_template_lib::{crypto::RistrettoPublicKeyBytes, prelude::NonFungibleAddress};
@@ -45,35 +42,38 @@ pub trait TransactionExecutor {
 pub struct ExecutionOutput {
     pub transaction: Transaction,
     pub result: ExecuteResult,
-    pub outputs: Vec<VersionedSubstateId>,
-    pub execution_time: Duration,
 }
 
 impl ExecutionOutput {
-    pub fn resolve_inputs(
-        &self,
-        inputs: IndexMap<VersionedSubstateId, Substate>,
-    ) -> Vec<VersionedSubstateIdLockIntent> {
+    pub fn resolve_inputs(&self, inputs: &IndexMap<SubstateId, Substate>) -> Vec<VersionedSubstateIdLockIntent> {
         if let Some(diff) = self.result.finalize.accept() {
             inputs
-                .into_iter()
-                .map(|(versioned_id, _)| {
-                    let lock_flag = if diff.down_iter().any(|(id, _)| *id == versioned_id.substate_id) {
+                .iter()
+                .map(|(substate_id, substate)| {
+                    let lock_flag = if diff.down_iter().any(|(id, _)| id == substate_id) {
                         // Update all inputs that were DOWNed to be write locked
-                        SubstateLockFlag::Write
+                        SubstateLockType::Write
                     } else {
                         // Any input not downed, gets a read lock
-                        SubstateLockFlag::Read
+                        SubstateLockType::Read
                     };
-                    VersionedSubstateIdLockIntent::new(versioned_id, lock_flag)
+                    VersionedSubstateIdLockIntent::new(
+                        VersionedSubstateId::new(substate_id.clone(), substate.version()),
+                        lock_flag,
+                    )
                 })
                 .collect()
         } else {
             // TODO: we might want to have a SubstateLockFlag::None for rejected transactions so that we still know the
             // shards involved but do not lock them. We dont actually lock anything for rejected transactions anyway.
             inputs
-                .into_iter()
-                .map(|(versioned_id, _)| VersionedSubstateIdLockIntent::new(versioned_id, SubstateLockFlag::Read))
+                .iter()
+                .map(|(substate_id, substate)| {
+                    VersionedSubstateIdLockIntent::new(
+                        VersionedSubstateId::new(substate_id.clone(), substate.version()),
+                        SubstateLockType::Read,
+                    )
+                })
                 .collect()
         }
     }
@@ -107,7 +107,6 @@ where TTemplateProvider: TemplateProvider<Template = LoadedTemplate>
         state_store: MemoryStateStore,
         virtual_substates: VirtualSubstates,
     ) -> Result<ExecutionOutput, Self::Error> {
-        let timer = Instant::now();
         // Include signature public key badges for all transaction signers in the initial auth scope
         // NOTE: we assume all signatures have already been validated.
         let initial_ownership_proofs = transaction
@@ -130,30 +129,19 @@ where TTemplateProvider: TemplateProvider<Template = LoadedTemplate>
             modules,
             self.network,
         );
-        let tx_id = transaction.hash();
-        let result = match processor.execute(transaction.clone()) {
-            Ok(result) => result,
-            Err(err) => ExecuteResult {
-                finalize: FinalizeResult::new_rejected(tx_id, RejectReason::ExecutionFailure(err.to_string())),
-            },
-        };
-
-        let outputs = result
-            .finalize
-            .result
-            .accept()
-            .map(|diff| {
-                diff.up_iter()
-                    .map(|(addr, substate)| VersionedSubstateId::new(addr.clone(), substate.version()))
-                    .collect::<Vec<_>>()
-            })
-            .unwrap_or_default();
-        Ok(ExecutionOutput {
-            transaction,
-            result,
-            outputs,
-            execution_time: timer.elapsed(),
-        })
+        let result = processor.execute(transaction.clone())?;
+        //     Ok(result) => result,
+        //     // TODO: This may occur due to an internal error (e.g. OOM, etc).
+        //     Err(err) => ExecuteResult {
+        //         finalize: FinalizeResult::new_rejected(
+        //             tx_id,
+        //             RejectReason::ExecutionFailure(format!("BUG: {err}")),
+        //         ),
+        //         execution_time: Duration::default(),
+        //     },
+        // };
+
+        Ok(ExecutionOutput { transaction, result })
     }
 }
 

applications/tari_dan_wallet_daemon/src/handlers/accounts.rs

@@ -947,11 +947,12 @@ pub async fn handle_transfer(
             .transaction_service()
             .submit_dry_run_transaction(transaction, required_inputs)
             .await?;
+        let finalize = execute_result.finalize;
         return Ok(AccountsTransferResponse {
             transaction_id,
-            fee: execute_result.fee_receipt.total_fees_paid,
-            fee_refunded: execute_result.fee_receipt.total_fee_payment - execute_result.fee_receipt.total_fees_paid,
-            result: execute_result,
+            fee: finalize.fee_receipt.total_fees_paid,
+            fee_refunded: finalize.fee_receipt.total_fee_payment - finalize.fee_receipt.total_fees_paid,
+            result: finalize,
         });
     }
 
@@ -1022,12 +1023,13 @@ pub async fn handle_confidential_transfer(
 
         if req.dry_run {
             let transaction_id = *transfer.transaction.id();
-            let finalize = transaction_service
+            let exec_result = transaction_service
                 .submit_dry_run_transaction(
                     transfer.transaction,
                     transfer.inputs.into_iter().map(Into::into).collect(),
                 )
                 .await?;
+            let finalize = exec_result.finalize;
             return Ok(ConfidentialTransferResponse {
                 transaction_id,
                 fee: finalize.fee_receipt.total_fees_paid,

applications/tari_dan_wallet_daemon/src/handlers/transaction.rs

@@ -8,12 +8,7 @@ use log::*;
 use tari_dan_app_utilities::json_encoding;
 use tari_dan_common_types::{optional::Optional, Epoch};
 use tari_dan_wallet_sdk::apis::{jwt::JrpcPermission, key_manager};
-use tari_engine_types::{
-    commit_result::ExecuteResult,
-    indexed_value::IndexedValue,
-    instruction::Instruction,
-    substate::SubstateId,
-};
+use tari_engine_types::{indexed_value::IndexedValue, instruction::Instruction, substate::SubstateId};
 use tari_template_lib::{args, args::Arg, models::Amount};
 use tari_transaction::Transaction;
 use tari_wallet_daemon_client::types::{
@@ -152,16 +147,16 @@ pub async fn handle_submit(
         transaction.hash()
     );
     if req.is_dry_run {
-        let finalize = context
+        let exec_result = context
             .transaction_service()
             .submit_dry_run_transaction(transaction, inputs.clone())
             .await?;
 
-        let json_result = json_encoding::encode_finalize_result_into_json(&finalize)?;
+        let json_result = json_encoding::encode_finalize_result_into_json(&exec_result.finalize)?;
 
         Ok(TransactionSubmitResponse {
-            transaction_id: finalize.transaction_hash.into_array().into(),
-            result: Some(ExecuteResult { finalize }),
+            transaction_id: exec_result.finalize.transaction_hash.into_array().into(),
+            result: Some(exec_result),
             json_result: Some(json_result),
             inputs,
         })

applications/tari_dan_wallet_daemon/src/services/transaction_service/handle.rs

@@ -2,7 +2,7 @@
 //   SPDX-License-Identifier: BSD-3-Clause
 
 use tari_dan_wallet_sdk::models::NewAccountInfo;
-use tari_engine_types::commit_result::FinalizeResult;
+use tari_engine_types::commit_result::ExecuteResult;
 use tari_transaction::{SubstateRequirement, Transaction, TransactionId};
 use tokio::sync::{mpsc, oneshot};
 
@@ -21,7 +21,7 @@ pub(super) enum TransactionServiceRequest {
     SubmitDryRunTransaction {
         transaction: Transaction,
         required_substates: Vec<SubstateRequirement>,
-        reply: Reply<Result<FinalizeResult, TransactionServiceError>>,
+        reply: Reply<Result<ExecuteResult, TransactionServiceError>>,
     },
 }
 
@@ -60,7 +60,7 @@ impl TransactionServiceHandle {
         &self,
         transaction: Transaction,
         required_substates: Vec<SubstateRequirement>,
-    ) -> Result<FinalizeResult, TransactionServiceError> {
+    ) -> Result<ExecuteResult, TransactionServiceError> {
         let (reply_tx, reply_rx) = oneshot::channel();
         self.sender
             .send(TransactionServiceRequest::SubmitDryRunTransaction {

applications/tari_dan_wallet_daemon/src/services/transaction_service/service.rs

@@ -11,6 +11,7 @@ use tari_dan_wallet_sdk::{
     storage::WalletStore,
     DanWalletSdk,
 };
+use tari_engine_types::commit_result::ExecuteResult;
 use tari_shutdown::ShutdownSignal;
 use tari_transaction::{SubstateRequirement, Transaction, TransactionId};
 use tokio::{
@@ -130,11 +131,15 @@ where
                         // Unlock all proofs related to the transaction
                         transaction_api.release_all_outputs_for_transaction(transaction_id)?;
 
+                        let finalize = finalized_transaction.finalize.ok_or_else(|| {
+                            TransactionServiceError::DryRunTransactionFailed {
+                                details: "Transaction was not finalized".to_string(),
+                            }
+                        });
                         reply
-                            .send(finalized_transaction.finalize.ok_or_else(|| {
-                                TransactionServiceError::DryRunTransactionFailed {
-                                    details: "Transaction was not finalized".to_string(),
-                                }
+                            .send(finalize.map(|finalize| ExecuteResult {
+                                finalize,
+                                execution_time: finalized_transaction.execution_time.unwrap_or_default(),
                             }))
                             .map_err(|_| TransactionServiceError::ServiceShutdown)?;
                     },

applications/tari_dan_wallet_web_ui/src/routes/AssetVault/Components/AddAccount.tsx

@@ -39,13 +39,15 @@ function AddAccount({ open, setOpen }: { open: boolean; setOpen: React.Dispatch<
   });
   const { mutateAsync: mutateAddAccount } = useAccountsCreate(accountFormState.accountName, null, null, false);
   const theme = useTheme();
+  const [isBusy, setIsBusy] = useState(false);
 
   const handleClose = () => {
     setOpen(false);
   };
 
-  const onSubmitAddAccount = () => {
-    mutateAddAccount(undefined, {
+  const onSubmitAddAccount = async () => {
+    setIsBusy(true);
+    await mutateAddAccount(undefined, {
       onSettled: () => {
         setAccountFormState({
           accountName: "",
@@ -54,6 +56,7 @@ function AddAccount({ open, setOpen }: { open: boolean; setOpen: React.Dispatch<
         queryClient.invalidateQueries(["accounts"]);
       },
     });
+    setIsBusy(false);
   };
 
   const onAccountChange = (e: React.ChangeEvent<HTMLInputElement>) => {
@@ -88,7 +91,7 @@ function AddAccount({ open, setOpen }: { open: boolean; setOpen: React.Dispatch<
             <Button variant="outlined" onClick={handleClose}>
               Cancel
             </Button>
-            <Button variant="contained" type="submit">
+            <Button variant="contained" type="submit" disabled={isBusy}>
               Add Account
             </Button>
           </Box>

applications/tari_dan_wallet_web_ui/src/routes/Transactions/TransactionDetails.tsx

@@ -122,23 +122,11 @@ export default function TransactionDetails() {
       } else {
         reason = txResult.Reject;
       }
-      const getReasonString = (x: RejectReason): string => {
-        if (typeof x === "string") {
-          return x;
-        } else if ("ShardsNotPledged" in x) {
-          return `ShardsNotPledged: ${x["ShardsNotPledged"]}`;
-        } else if ("ExecutionFailure" in x) {
-          return `ExecutionFailure: ${x["ExecutionFailure"]}`;
-        } else if ("ShardPledgedToAnotherPayload" in x) {
-          return `ShardPledgedToAnotherPayload: ${x["ShardPledgedToAnotherPayload"]}`;
-        } else if ("ShardRejected" in x) {
-          return `ShardRejected: ${x["ShardRejected"]}`;
-        } else if ("FeesNotPaid" in x) {
-          return `FeesNotPaid: ${x["FeesNotPaid"]}`;
-        }
-        return "Unknown reason";
-      };
-      return getReasonString(reason);
+      if (typeof reason === "string") {
+        return reason;
+      } else {
+        return JSON.stringify(reason);
+      }
     };
 
     if (data.status === "Rejected" || data.status === "InvalidTransaction") {

applications/tari_dan_wallet_web_ui/src/utils/json_rpc.tsx

@@ -83,7 +83,12 @@ import { AccountGetDefaultRequest, TemplatesGetRequest, WalletDaemonClient } fro
 let clientInstance: WalletDaemonClient | null = null;
 let pendingClientInstance: Promise<WalletDaemonClient> | null = null;
 let outerAddress: URL | null = null;
-const DEFAULT_WALLET_ADDRESS = new URL(import.meta.env.VITE_DAEMON_JRPC_ADDRESS || "http://localhost:9000");
+const DEFAULT_WALLET_ADDRESS = new URL(
+  import.meta.env.VITE_DAEMON_JRPC_ADDRESS ||
+    import.meta.env.VITE_JSON_RPC_ADDRESS ||
+    import.meta.env.VITE_JRPC_ADDRESS ||
+    "http://localhost:9000",
+);
 
 export async function getClientAddress(): Promise<URL> {
   try {

applications/tari_indexer/src/event_scanner.rs

@@ -402,7 +402,7 @@ impl EventScanner {
     fn extract_transactions_from_blocks(&self, blocks: Vec<Block>) -> Vec<TransactionMetadata> {
         blocks
             .iter()
-            .flat_map(|b| b.all_accepted_transactions_ids().map(|id| (id, b.timestamp())))
+            .flat_map(|b| b.all_committing_transactions_ids().map(|id| (id, b.timestamp())))
             .map(|(transaction_id, timestamp)| TransactionMetadata {
                 transaction_id: *transaction_id,
                 timestamp,