fix(wallet)!: private derived tag and optimised* sync protocol

applications/tari_indexer/Cargo.toml

@@ -34,12 +34,14 @@ tari_template_lib = { workspace = true }
 tari_template_manager = { workspace = true }
 tari_transaction = { workspace = true }
 tari_validator_node_rpc = { workspace = true }
+tari_ootle_wallet_sdk = { workspace = true }
 
 anyhow = { workspace = true }
 axum = { workspace = true }
 async-graphql = { workspace = true }
 async-graphql-axum = { workspace = true }
 axum-jrpc = { workspace = true, features = ["anyhow_error"] }
+bincode = { workspace = true, features = ["serde"] }
 clap = { workspace = true, features = ["derive", "env"] }
 config = { workspace = true }
 diesel = { workspace = true, default-features = false, features = [

applications/tari_indexer/log4rs_sample.yml

@@ -36,7 +36,7 @@ appenders:
         count: 5
         pattern: "{{log_dir}}/log/indexer/network.{}.log"
     encoder:
-      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [Thread:{I}] {l:5} {m}// {f}:{L}{n}"
+      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] {l:5} {m}// {f}:{L}{n}"
 
   # An appender named "ootle" that writes to a file with a custom pattern encoder
   ootle:
@@ -53,7 +53,7 @@ appenders:
         count: 5
         pattern: "{{log_dir}}/log/indexer/ootle.{}.log"
     encoder:
-      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [{X(node-public-key)},{X(node-id)}] {l:5} {m} // {f}:{L}{n}"
+      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] {l:5} {m} // {f}:{L}{n}"
 
   # An appender named "ootle" that writes to a file with a custom pattern encoder
   json_rpc:
@@ -70,7 +70,7 @@ appenders:
         count: 5
         pattern: "{{log_dir}}/log/indexer/json_rpc.{}.log"
     encoder:
-      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [{X(node-public-key)},{X(node-id)}] {l:5} {m} // {f}:{L}{n}"
+      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] {l:5} {m} // {f}:{L}{n}"
 
   # An appender named "libp2p" that writes to a file with a custom pattern encoder
   libp2p:

applications/tari_indexer/src/bootstrap.rs

@@ -66,7 +66,10 @@ use crate::{
     network_state_sync::NetworkWideStateSyncConfig,
     storage_sqlite::{
         models::Key,
-        store_factory::{IndexerStore, IndexerStoreReadTransaction, IndexerStoreWriteTransaction, SqliteIndexerStore},
+        IndexerStore,
+        IndexerStoreReadTransaction,
+        IndexerStoreWriteTransaction,
+        SqliteIndexerStore,
     },
     ApplicationConfig,
     IndexerEpochManagerSpec,

applications/tari_indexer/src/event_manager.rs

@@ -26,7 +26,7 @@ use log::*;
 use tari_engine_types::{events::Event, substate::SubstateId};
 use tari_template_lib::{models::Metadata, types::Hash};
 
-use crate::storage_sqlite::store_factory::{IndexerStore, IndexerStoreReadTransaction, SqliteIndexerStore};
+use crate::storage_sqlite::{IndexerStore, IndexerStoreReadTransaction, SqliteIndexerStore};
 
 const LOG_TARGET: &str = "tari::indexer::event_manager";
 

applications/tari_indexer/src/json_rpc/handlers.rs

@@ -54,6 +54,8 @@ use tari_indexer_client::types::{
     GetTemplateDefinitionResponse,
     GetTransactionResultRequest,
     GetTransactionResultResponse,
+    GetUnspentUtxosRequest,
+    GetUnspentUtxosResponse,
     GetUtxoUpdatesRequest,
     GetUtxoUpdatesResponse,
     IndexerReadyResponse,
@@ -66,7 +68,6 @@ use tari_indexer_client::types::{
     ListSubstatesResponse,
     ListTemplatesRequest,
     ListTemplatesResponse,
-    NonFungibleSubstate,
     SubmitTransactionRequest,
     SubmitTransactionResponse,
     TemplateMetadata,
@@ -86,6 +87,7 @@ use tari_ootle_storage::{
     time::{PrimitiveDateTime, UtcDateTime},
 };
 use tari_ootle_storage_sqlite::global::SqliteGlobalDbAdapter;
+use tari_ootle_wallet_sdk::models::{UtxoStateUpdateSet, UtxoUpdateSet};
 use tari_template_manager::{
     implementation::TemplateManager,
     interface::{TemplateExecutable, TemplateManagerError},
@@ -97,7 +99,7 @@ use crate::{
     dry_run::processor::DryRunTransactionProcessor,
     json_rpc::error::internal_error,
     network_client::NetworkClientError,
-    storage_sqlite::store_factory::SqliteIndexerStore,
+    storage_sqlite::SqliteIndexerStore,
     substate_manager::SubstateManager,
     transaction_manager::{error::TransactionManagerError, TransactionManager},
 };
@@ -433,19 +435,12 @@ impl JsonRpcHandlers {
             )
         })?;
 
-        let res = self
+        let non_fungibles = self
             .substate_manager
             .get_non_fungibles_by_resource_address(request.address, limit, offset)
             .map_err(|e| Self::internal_error(answer_id, format!("Error getting non fungibles: {}", e)))?;
         Ok(JsonRpcResponse::success(answer_id, GetNonFungiblesResponse {
-            non_fungibles: res
-                .into_iter()
-                .map(|v| NonFungibleSubstate {
-                    address: v.address,
-                    version: v.version,
-                    substate: v.substate,
-                })
-                .collect(),
+            non_fungibles,
         }))
     }
 
@@ -496,18 +491,12 @@ impl JsonRpcHandlers {
             ));
         }
 
-        let mut utxo_updates = Vec::new();
+        let mut utxo_updates = HashMap::new();
         let mut per_shard_high_watermark = Vec::with_capacity(req.shard_state_versions.len());
         for (shard, state_version) in req.shard_state_versions {
-            let (max_version, updates) = self
+            let (max_state_version, updates) = self
                 .substate_manager
-                .get_utxo_updates(
-                    req.resource_address,
-                    shard,
-                    state_version,
-                    &req.filter_tag_bytes,
-                    req.per_shard_limit,
-                )
+                .get_utxo_updates(req.resource_address, shard, state_version, req.per_shard_limit)
                 .map_err(|e| {
                     Self::internal_error(
                         answer_id,
@@ -517,19 +506,72 @@ impl JsonRpcHandlers {
                         ),
                     )
                 })?;
-            if max_version.as_u64() > 0 {
+
+            // TODO: this is on the hot path, figure out a better way to let the client know the max shard versions
+            // without sending it each time
+            let current_tip_version = self
+                .substate_manager
+                .get_max_state_version(&req.resource_address, shard)
+                .map_err(|e| {
+                    Self::internal_error(
+                        answer_id,
+                        format!(
+                            "Error getting max state version for resource_address {}, shard {}: {}",
+                            req.resource_address, shard, e
+                        ),
+                    )
+                })?;
+            if current_tip_version.as_u64() > 0 {
                 // Save a little over the wire initially by not sending 0 watermarks
-                per_shard_high_watermark.push((shard, max_version));
+                per_shard_high_watermark.push((shard, current_tip_version));
+            }
+            if !updates.is_empty() {
+                utxo_updates.insert(shard, UtxoStateUpdateSet {
+                    updates,
+                    max_state_version,
+                });
             }
-            utxo_updates.extend(updates);
         }
 
         Ok(JsonRpcResponse::success(answer_id, GetUtxoUpdatesResponse {
-            utxo_updates,
-            per_shard_high_watermark,
+            updates: UtxoUpdateSet {
+                shard_updates: utxo_updates,
+                per_shard_high_watermark,
+            },
         }))
     }
 
+    pub async fn get_unspent_utxos(&self, value: JsonRpcExtractor) -> JrpcResult {
+        let answer_id = value.get_answer_id();
+        let req: GetUnspentUtxosRequest = value.parse_params()?;
+        if req.tag_and_nonce_pairs.len() > 1000 {
+            return Err(JsonRpcResponse::error(
+                answer_id,
+                JsonRpcError::new(
+                    JsonRpcErrorReason::InvalidParams,
+                    "cannot query more than 1000 UTXOs".to_string(),
+                    Value::Null,
+                ),
+            ));
+        }
+        let utxos = self
+            .substate_manager
+            .get_unspent_utxos(&req.resource_address, &req.tag_and_nonce_pairs)
+            .map_err(|e| {
+                Self::internal_error(
+                    answer_id,
+                    format!(
+                        "Error getting UTXOs for resource_address {}, with {} tag/nonce pair(s): {}",
+                        req.resource_address,
+                        req.tag_and_nonce_pairs.len(),
+                        e
+                    ),
+                )
+            })?;
+
+        Ok(JsonRpcResponse::success(answer_id, GetUnspentUtxosResponse { utxos }))
+    }
+
     pub async fn submit_transaction(&self, value: JsonRpcExtractor) -> JrpcResult {
         let answer_id = value.get_answer_id();
         let request: SubmitTransactionRequest = value.parse_params()?;

applications/tari_indexer/src/json_rpc/server.rs

@@ -78,6 +78,7 @@ async fn handler(Extension(handlers): Extension<Arc<JsonRpcHandlers>>, value: Js
         "inspect_substate" => handlers.inspect_substate(value).await,
         "get_non_fungibles" => handlers.get_non_fungibles(value).await,
         "get_utxo_updates" => handlers.get_utxo_updates(value).await,
+        "get_unspent_utxos" => handlers.get_unspent_utxos(value).await,
 
         // Transactions
         "submit_transaction" => handlers.submit_transaction(value).await,

applications/tari_indexer/src/network_state_sync/block_scanner.rs

@@ -19,7 +19,10 @@ use crate::{
     block_data::BlockData,
     storage_sqlite::{
         models::{NewScannedBlockId, NewSubstate},
-        store_factory::{IndexerStore, IndexerStoreReadTransaction, IndexerStoreWriteTransaction, SqliteIndexerStore},
+        IndexerStore,
+        IndexerStoreReadTransaction,
+        IndexerStoreWriteTransaction,
+        SqliteIndexerStore,
     },
 };
 

applications/tari_indexer/src/network_state_sync/worker.rs

@@ -18,9 +18,6 @@ use tari_ootle_common_types::{
     PeerAddress,
     ShardGroup,
     StateVersion,
-    UtxoSpent,
-    UtxoUnspent,
-    UtxoUpdate,
     VotePower,
 };
 use tari_ootle_p2p::{proto::rpc, TariMessagingSpec};
@@ -43,14 +40,12 @@ use crate::{
         sync_progress::SyncProgress,
     },
     storage_sqlite::{
-        models::Key,
-        store_factory::{
-            IndexerStore,
-            IndexerStoreReadTransaction,
-            IndexerStoreWriteTransaction,
-            SqliteIndexerStore,
-            SqliteStoreWriteTransaction,
-        },
+        models::{Key, UtxoSpent, UtxoUnspent, UtxoUpdateRecord},
+        IndexerStore,
+        IndexerStoreReadTransaction,
+        IndexerStoreWriteTransaction,
+        SqliteIndexerStore,
+        SqliteStoreWriteTransaction,
     },
 };
 
@@ -327,7 +322,7 @@ impl NetworkWideStateSync {
         shard: Shard,
         sync_plan_mut: &mut SyncPlan,
         update_buf: &mut Vec<(Epoch, SubstateUpdateProof)>,
-        utxos_buf: &mut Vec<UtxoUpdate>,
+        utxos_buf: &mut Vec<UtxoUpdateRecord>,
         transactions_buf: &mut Vec<TransactionReceipt>,
         shard_group: ShardGroup,
         session: &mut ValidatorRpcSession,
@@ -452,20 +447,24 @@ fn extend_bufs_from_substate_update(
     msg_epoch: Epoch,
     update_buf: &mut Vec<(Epoch, SubstateUpdateProof)>,
     templates_buf: &mut Vec<TemplateChange>,
-    utxos_buf: &mut Vec<UtxoUpdate>,
+    utxos_buf: &mut Vec<UtxoUpdateRecord>,
     transactions_buf: &mut Vec<TransactionReceipt>,
 ) -> Result<(), NetworkStateSyncError> {
     match &update {
         SubstateUpdateProof::Create(create) => match create.substate.value().value() {
             Some(SubstateValue::Utxo(utxo)) => {
                 if let Some(address) = create.substate.substate_id().as_utxo_address() {
-                    utxos_buf.push(UtxoUpdate::Unspent(UtxoUnspent {
-                        address,
-                        version: update.version(),
-                        shard,
-                        state_version,
-                        utxo: utxo.clone(),
-                    }));
+                    let is_frozen = utxo.is_frozen();
+                    if let Some(ref output) = utxo.output {
+                        utxos_buf.push(UtxoUpdateRecord::Unspent(UtxoUnspent {
+                            address,
+                            version: update.version(),
+                            shard,
+                            state_version,
+                            utxo_output: output.clone(),
+                            is_frozen,
+                        }));
+                    }
                 } else {
                     warn!(target: LOG_TARGET, "⚠️ NEVER HAPPEN: Received UTXO substate with invalid address: {}", create.substate.substate_id());
                 };
@@ -505,7 +504,7 @@ fn extend_bufs_from_substate_update(
                 });
             },
             SubstateId::Utxo(address) => {
-                utxos_buf.push(UtxoUpdate::Spent(UtxoSpent {
+                utxos_buf.push(UtxoUpdateRecord::Spent(UtxoSpent {
                     address: address.clone(),
                     shard,
                     version: update.version(),

applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql

@@ -119,19 +119,20 @@ CREATE UNIQUE INDEX epoch_checkpoints_uniq_epoch_shard_group ON epoch_checkpoint
 create table utxos
 (
     id               integer   not NULL primary key AUTOINCREMENT,
-    address          text      not NULL,
+    commitment       text      not NULL,
+    public_nonce     text      not NULL,
     version          int       not NULL,
     resource_address text      not NULL,
     shard            int       not NULL,
     state_version    bigint    not NULL,
-    output           text      NULL,
-    utxo_tag_byte    int       NULL,
+    output           blob      NULL,
+    utxo_tag         int       not NULL,
     is_spent         boolean   not NULL,
     is_burnt         boolean   not NULL,
     is_frozen        boolean   not NULL,
     created_at       timestamp not null default current_timestamp
 );
 
-CREATE INDEX utxos_shard_tag_resource_state_version_idx
-    ON utxos (shard, utxo_tag_byte, resource_address, state_version);
+CREATE INDEX utxos_resource_state_version_shard_idx ON utxos (resource_address, state_version, shard);
+CREATE UNIQUE INDEX utxos_resource_public_nonce_utxo_tag_uniq_partial ON utxos (resource_address, public_nonce, utxo_tag) WHERE is_spent = false;