feat!: impl epoch birthday for utxo scanning

applications/tari_indexer/src/config.rs

@@ -80,6 +80,10 @@ impl ApplicationConfig {
     pub fn state_db_path(&self) -> PathBuf {
         self.to_data_dir().join("state.db")
     }
+
+    pub fn global_db_path(&self) -> PathBuf {
+        self.to_data_dir().join("global_storage.sqlite")
+    }
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]

applications/tari_indexer/src/lib.rs

@@ -79,7 +79,7 @@ pub async fn run_indexer(config: ApplicationConfig, mut shutdown_signal: Shutdow
     info!(target: LOG_TARGET, "Starting indexer node on network {}", config.network);
     let keypair = setup_keypair_prompt(config.to_identity_file_path(), true)?;
 
-    let db_factory = SqliteDbFactory::new(config.indexer.data_dir.clone());
+    let db_factory = SqliteDbFactory::new(config.global_db_path());
     db_factory
         .migrate()
         .map_err(|e| ExitError::new(ExitCode::DatabaseError, e))?;

applications/tari_indexer/src/network_state_sync/worker.rs

@@ -412,7 +412,7 @@ impl NetworkWideStateSync {
                 // TODO: this is not currently used. Consider removing. 
                 tx.batch_insert_substate_transitions(shard, state_version, update_buf.drain(..))?;
                 debug!(target: LOG_TARGET, "✅ Committing {} UTXOs for shard {shard} (epoch: {msg_epoch})", utxos_buf.len());
-                tx.batch_insert_utxo_updates(utxos_buf.drain(..))?;
+                tx.batch_insert_utxo_updates(msg_epoch, utxos_buf.drain(..))?;
                 // TODO: there are many ways to do this. This is probably not the best way. But this allows wallet to query for validator fee pool values since
                 // block sync does not sync validator fee pools (due to block diffs being removed on block commit).
                 for substate_data in validator_fee_pools_buf.drain(..) {

applications/tari_indexer/src/rest_api/handlers/substates.rs

@@ -67,6 +67,16 @@ pub async fn get_substate(
     Path(substate_id): Path<SubstateId>,
     Query(req): Query<GetSubstateRequest>,
 ) -> HandlerResult<Json<GetSubstateResponse>> {
+    if !context
+        .epoch_manager()
+        .is_initial_scanning_complete()
+        .await
+        .map_err(ErrorResponse::anyhow)?
+    {
+        return Err(ErrorResponse::service_unavailable(
+            "Indexer is still syncing. Please try again later.",
+        ));
+    }
     let maybe_substate = context
         .substate_manager()
         .get_substate(&substate_id, req.version)

applications/tari_indexer/src/rest_api/handlers/transactions.rs

@@ -68,13 +68,17 @@ pub async fn submit_transaction(
         .submit_transaction(transaction)
         .await
         .map_err(|e| match e {
-            TransactionManagerError::NetworkClientError(NetworkClientError::AllValidatorsFailed { .. }) => {
+            TransactionManagerError::NetworkClientError(NetworkClientError::AllValidatorsFailed { .. }) |
+            TransactionManagerError::NetworkClientError(NetworkClientError::NoCommitteeMembers) => {
                 ErrorResponse::service_unavailable(format!("All validators failed: {}", e))
             },
             TransactionManagerError::InvalidTransaction {
                 transaction_id,
                 details,
             } => ErrorResponse::bad_request(format!("Transaction {} is invalid: {}", transaction_id, details)),
+            TransactionManagerError::NetworkClientError(NetworkClientError::NoInputsProvided) => {
+                ErrorResponse::bad_request("Transaction has no inputs".to_string())
+            },
             e => ErrorResponse::anyhow(e),
         })?;
 

applications/tari_indexer/src/rest_api/streaming/utxo_stream.rs

@@ -12,7 +12,7 @@ use futures::Stream;
 use log::*;
 use tari_indexer_client::{protobuf, types::GetUtxoUpdatesRequest};
 use tari_ootle_common_types::{shard::Shard, StateVersion};
-use tari_ootle_wallet_sdk::models::WalletUtxoUpdate;
+use tari_ootle_wallet_sdk::models::{UtxoStateUpdateSet, WalletUtxoUpdate};
 
 use crate::{
     rest_api::{encoder::Encoder, error::ErrorResponse, streaming::encoding::MimeTypeEncoder},
@@ -98,8 +98,13 @@ impl UtxoUpdateStream {
     }
 
     pub fn next_batch(&mut self, shard: Shard, state_version: StateVersion) -> anyhow::Result<bool> {
-        let (updates_state_version, updates) = self.substate_manager.get_utxo_updates(
+        let UtxoStateUpdateSet {
+            updates,
+            max_state_version,
+            max_epoch,
+        } = self.substate_manager.get_utxo_updates(
             self.request.resource_address,
+            self.request.from_epoch,
             shard,
             state_version,
             self.request.unspent_only,
@@ -113,16 +118,13 @@ impl UtxoUpdateStream {
         }
         debug!(
             target: LOG_TARGET,
-            "Received {} updates for shard {}, max_state_version {} -> {}",
+            "Received {} updates for shard {shard}, max_epoch = {max_epoch}, max_state_version {max_state_version} -> {high_watermark_state_version}",
             updates.len(),
-            shard,
-            updates_state_version,
-            high_watermark_state_version
         );
         self.pending_updates = Some(PendingUpdates {
             sos_emitted: false,
             shard,
-            updates_state_version,
+            updates_state_version: max_state_version,
             high_watermark_state_version,
             updates,
             index: 0,

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

@@ -136,12 +136,13 @@ create table utxos
     state_version    bigint    not NULL,
     output           blob      NULL,
     utxo_tag         int       not NULL,
+    epoch            bigint    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_resource_state_version_shard_idx ON utxos (resource_address, state_version, shard);
+CREATE INDEX utxos_resource_state_version_shard_epoch_idx ON utxos (resource_address, state_version, shard, epoch);
 CREATE UNIQUE INDEX utxos_resource_public_nonce_utxo_tag_uniq_partial ON utxos (resource_address, public_nonce, utxo_tag) WHERE is_spent = false;
 

applications/tari_indexer/src/storage_sqlite/models/utxo.rs

@@ -17,6 +17,7 @@ use crate::storage_sqlite::{schema::utxos, serialization::deserialize_bincode};
 #[derive(AsChangeset, Default)]
 #[diesel(table_name = utxos)]
 pub(crate) struct UtxoRecordUpdate {
+    pub epoch: Option<i64>,
     pub version: Option<i32>,
     pub output: Option<Option<Vec<u8>>>,
     pub state_version: Option<i64>,
@@ -36,6 +37,7 @@ pub(crate) struct UtxoRecordInsert {
     pub state_version: i64,
     pub output: Option<Vec<u8>>,
     pub utxo_tag: i32,
+    pub epoch: i64,
     pub is_spent: bool,
     pub is_burnt: bool,
     pub is_frozen: bool,
@@ -52,6 +54,7 @@ pub(crate) struct UtxoRecord {
     pub state_version: i64,
     pub output: Option<Vec<u8>>,
     pub _utxo_tag: i32,
+    pub epoch: i64,
     pub _is_spent: bool,
     pub is_burnt: bool,
     pub is_frozen: bool,

applications/tari_indexer/src/storage_sqlite/reader.rs

@@ -35,7 +35,7 @@ use tari_ootle_common_types::{
 };
 use tari_ootle_storage::{time::PrimitiveDateTime, Ordering, StorageError};
 use tari_ootle_storage_sqlite::SqliteTransaction;
-use tari_ootle_wallet_sdk::models::WalletUtxoUpdate;
+use tari_ootle_wallet_sdk::models::UtxoStateUpdateSet;
 use tari_template_lib::{
     models::{ResourceAddress, UtxoId},
     prelude::{RistrettoPublicKeyBytes, TemplateAddress},
@@ -519,17 +519,19 @@ impl IndexerStoreReadTransaction for SqliteStoreReadTransaction<'_> {
     fn utxos_get_updates(
         &mut self,
         resource_address: ResourceAddress,
+        from_epoch: Epoch,
         shard: Shard,
         from_state_version: StateVersion,
         unspent_only: bool,
         limit: u32,
-    ) -> Result<(StateVersion, Vec<WalletUtxoUpdate>), StorageError> {
+    ) -> Result<UtxoStateUpdateSet, StorageError> {
         const OPERATION: &str = "get_utxo_updates";
         use crate::storage_sqlite::schema::utxos;
 
         let mut query = utxos::table
             .filter(utxos::resource_address.eq(resource_address.to_string()))
             .filter(utxos::state_version.gt(from_state_version.as_u64() as i64))
+            .filter(utxos::epoch.ge(from_epoch.as_u64() as i64))
             .filter(utxos::shard.eq(shard.as_u32() as i32))
             .limit(i64::from(limit))
             .order_by(utxos::state_version.asc())
@@ -548,16 +550,23 @@ impl IndexerStoreReadTransaction for SqliteStoreReadTransaction<'_> {
 
         let mut updates = Vec::new();
         let mut max_state_version = StateVersion::zero();
+        let mut max_epoch = Epoch::zero();
         for row in rows {
             let row = row.map_err(|e| StorageError::QueryError {
                 reason: format!("{OPERATION}: {}", e),
             })?;
+            let epoch = Epoch(row.epoch as u64);
             let (state_version, update) = row.try_convert_to_update()?;
             max_state_version = max_state_version.max(state_version);
+            max_epoch = max_epoch.max(epoch);
             updates.push(update);
         }
 
-        Ok((max_state_version, updates))
+        Ok(UtxoStateUpdateSet {
+            updates,
+            max_state_version,
+            max_epoch,
+        })
     }
 
     fn utxos_list(

applications/tari_indexer/src/storage_sqlite/schema.rs

@@ -108,6 +108,7 @@ diesel::table! {
         state_version -> BigInt,
         output -> Nullable<Binary>,
         utxo_tag -> Integer,
+        epoch -> BigInt,
         is_spent -> Bool,
         is_burnt -> Bool,
         is_frozen -> Bool,

applications/tari_indexer/src/storage_sqlite/writer.rs

@@ -116,6 +116,7 @@ impl IndexerStoreWriteTransaction for SqliteStoreWriteTransaction<'_> {
 
     fn batch_insert_utxo_updates<I: IntoIterator<Item = UtxoUpdateRecord>>(
         &mut self,
+        epoch: Epoch,
         updates: I,
     ) -> Result<(), StorageError> {
         const OPERATION: &str = "batch_insert_utxo_updates";
@@ -136,6 +137,7 @@ impl IndexerStoreWriteTransaction for SqliteStoreWriteTransaction<'_> {
                         resource_address,
                         state_version: unspent.state_version.as_u64() as i64,
                         utxo_tag: unspent.utxo_output.tag.value() as i32,
+                        epoch: epoch.as_u64() as i64,
                         is_spent: false,
                         is_burnt: false,
                         is_frozen: unspent.is_frozen,
@@ -150,6 +152,7 @@ impl IndexerStoreWriteTransaction for SqliteStoreWriteTransaction<'_> {
                     let resource_address = spent.address.resource_address().to_string();
                     let commitment = spent.address.id().to_commitment_hex_string();
                     let update = UtxoRecordUpdate {
+                        epoch: Some(epoch.as_u64() as i64),
                         version: Some(spent.version as i32),
                         // Prune the UTXO data for spent outputs
                         output: Some(None),

applications/tari_indexer/src/store.rs

@@ -18,7 +18,7 @@ use tari_ootle_storage::{
     Ordering,
     StorageError,
 };
-use tari_ootle_wallet_sdk::models::WalletUtxoUpdate;
+use tari_ootle_wallet_sdk::models::UtxoStateUpdateSet;
 use tari_template_lib::{
     models::{ResourceAddress, UtxoId},
     prelude::RistrettoPublicKeyBytes,
@@ -138,17 +138,15 @@ pub trait IndexerStoreReadTransaction {
     ) -> Result<StateVersion, StorageError>;
 
     /// Get UTXO updates for a given resource address and shard, starting from a specific state version.
-    ///
-    /// Returns a tuple containing the maximum returned state version, and a vector of UTXO
-    /// updates.
     fn utxos_get_updates(
         &mut self,
         resource_address: ResourceAddress,
+        from_epoch: Epoch,
         shard: Shard,
         from_state_version: StateVersion,
         unspents_only: bool,
         limit: u32,
-    ) -> Result<(StateVersion, Vec<WalletUtxoUpdate>), StorageError>;
+    ) -> Result<UtxoStateUpdateSet, StorageError>;
 
     fn utxos_list(
         &mut self,
@@ -176,6 +174,7 @@ pub trait IndexerStoreWriteTransaction {
     ) -> Result<(), StorageError>;
     fn batch_insert_utxo_updates<I: IntoIterator<Item = UtxoUpdateRecord>>(
         &mut self,
+        epoch: Epoch,
         updates: I,
     ) -> Result<(), StorageError>;
     fn upsert_substate(&mut self, substate: &SubstateData) -> Result<(), StorageError>;

applications/tari_indexer/src/substate_manager.rs

@@ -34,11 +34,12 @@ use tari_indexer_lib::substate_scanner::SubstateScanner;
 use tari_ootle_common_types::{
     shard::Shard,
     substate_type::SubstateType,
+    Epoch,
     PeerAddress,
     StateVersion,
     VersionedSubstateIdRef,
 };
-use tari_ootle_wallet_sdk::models::WalletUtxoUpdate;
+use tari_ootle_wallet_sdk::models::UtxoStateUpdateSet;
 use tari_template_lib::{
     models::{ResourceAddress, UtxoId},
     types::{
@@ -107,13 +108,21 @@ impl SubstateManager {
     pub fn get_utxo_updates(
         &self,
         resource_address: ResourceAddress,
+        from_epoch: Epoch,
         shard: Shard,
         from_state_version: StateVersion,
         unspent_only: bool,
         limit: u32,
-    ) -> Result<(StateVersion, Vec<WalletUtxoUpdate>), anyhow::Error> {
+    ) -> Result<UtxoStateUpdateSet, anyhow::Error> {
         let updates = self.substate_store.with_read_tx(|tx| {
-            tx.utxos_get_updates(resource_address, shard, from_state_version, unspent_only, limit)
+            tx.utxos_get_updates(
+                resource_address,
+                from_epoch,
+                shard,
+                from_state_version,
+                unspent_only,
+                limit,
+            )
         })?;
         Ok(updates)
     }

applications/tari_validator_node/src/config.rs

@@ -131,6 +131,10 @@ impl ValidatorNodeConfig {
         //     self.database.sqlite.path = self.data_dir.as_ref().join(&self.database.sqlite.path);
         // }
     }
+
+    pub fn get_global_db_path(&self) -> PathBuf {
+        self.data_dir.join("global_storage.sqlite")
+    }
 }
 
 impl Default for ValidatorNodeConfig {

applications/tari_validator_node/src/lib.rs

@@ -92,7 +92,7 @@ pub async fn run_validator_node(
 ) -> Result<(), anyhow::Error> {
     info!(target: LOG_TARGET, "Starting validator node on network {}", config.network);
 
-    let db_factory = SqliteDbFactory::new(config.validator_node.data_dir.clone());
+    let db_factory = SqliteDbFactory::new(config.validator_node.get_global_db_path());
     db_factory
         .migrate()
         .map_err(|e| ExitError::new(ExitCode::DatabaseError, e))?;

applications/tari_validator_node/src/p2p/rpc/block_sync_task.rs

@@ -12,7 +12,7 @@ use tari_ootle_p2p::{
     proto::rpc::{sync_blocks_response::SyncData, QuorumCertificates, SyncBlocksResponse},
 };
 use tari_ootle_storage::{
-    consensus_models::{Block, SubstateCreatedProof, SubstateUpdateProof, TransactionRecord},
+    consensus_models::{Block, SubstateCreate, SubstateUpdateProof, TransactionRecord},
     StateStore,
     StateStoreReadTransaction,
     StorageError,
@@ -29,7 +29,7 @@ struct BlockData {
     qcs: Vec<ProposalCertificate>,
     substates: Vec<SubstateUpdateProof>,
     transactions: Vec<TransactionRecord>,
-    transaction_receipts: Vec<SubstateCreatedProof>,
+    transaction_receipts: Vec<SubstateCreate>,
 }
 type BlockBuffer = Vec<BlockData>;
 

applications/tari_walletd/src/lib.rs

@@ -33,10 +33,11 @@ use std::{fs, panic, pin, process};
 
 use log::*;
 use tari_common_types::seeds::seed_words::SeedWords;
-use tari_ootle_common_types::{optional::Optional, NumPreshards};
+use tari_ootle_common_types::{optional::Optional, Network, NumPreshards};
 use tari_ootle_wallet_sdk::{
     apis::config::{ConfigApi, ConfigKey},
     cipher_seed::CipherSeedRestore,
+    models::EpochBirthday,
     WalletSdk as Sdk,
     WalletSdkConfig,
 };
@@ -92,10 +93,12 @@ pub async fn run_tari_ootle_walletd(
 
     // trigger account scanning if needed
     if needs_seed_recovery {
+        let cipher_seed_birthday = wallet_sdk.key_manager_api().get_cipher_seed_birthday_epoch()?;
         let scanner = AccountRecoveryService::new(
             wallet_sdk.clone(),
             services.account_monitor_handle.clone(),
             config.ootle_wallet_daemon.recovery_abandon_count,
+            cipher_seed_birthday,
         );
         let shutdown_signal = shutdown_signal.clone();
         tokio::spawn(async move {
@@ -193,6 +196,19 @@ pub fn initialize_wallet_sdk(config: &ApplicationConfig, store: SqliteWalletStor
         config.ootle_wallet_daemon.indexer_api_url.clone()
     };
     let indexer = IndexerRestApiNetworkInterface::new(indexer_endpoint);
-    let sdk = WalletSdk::initialize(store, indexer, sdk_config)?;
+    let birthday = get_epoch_birthday(sdk_config.network);
+    let sdk = WalletSdk::initialize(store, indexer, sdk_config, birthday)?;
     Ok(sdk)
 }
+
+const fn get_epoch_birthday(network: Network) -> EpochBirthday {
+    // TODO: set the zero epoch time for each network according to actual zero epoch time
+    match network {
+        Network::MainNet => EpochBirthday::far_future(),
+        Network::StageNet => EpochBirthday::far_future(),
+        Network::NextNet => EpochBirthday::far_future(),
+        Network::LocalNet => EpochBirthday::far_future(),
+        Network::Igor => EpochBirthday::far_future(),
+        Network::Esmeralda => EpochBirthday::far_future(),
+    }
+}