fix: fix fee claiming, indexer syncs VN fee pool substates

applications/tari_indexer/src/network_state_sync/block_scanner.rs

@@ -4,21 +4,16 @@
 use futures::StreamExt;
 use log::*;
 use tari_consensus_types::BlockId;
-use tari_engine_types::substate::SubstateValue;
 use tari_epoch_manager::{service::EpochManagerHandle, EpochManagerReader};
 use tari_ootle_common_types::{committee::Committee, Epoch, PeerAddress, ShardGroup};
 use tari_ootle_p2p::{proto, proto::rpc::SyncBlocksRequest};
-use tari_ootle_storage::{
-    consensus_models::{Block, SubstateUpdateProof},
-    time::{OffsetDateTime, PrimitiveDateTime},
-};
-use tari_template_lib::types::TemplateAddress;
+use tari_ootle_storage::consensus_models::{Block, SubstateUpdateProof};
 use tari_validator_node_rpc::client::{TariValidatorNodeRpcClientFactory, ValidatorNodeClientFactory};
 
 use crate::{
     block_data::BlockData,
     storage_sqlite::{
-        models::{NewScannedBlockId, NewSubstate},
+        models::NewScannedBlockId,
         IndexerStore,
         IndexerStoreReadTransaction,
         IndexerStoreWriteTransaction,
@@ -107,12 +102,19 @@ impl BlockScanner {
 
             count += new_blocks.len();
             for block_data in new_blocks {
-                let timestamp = unix_epoch_to_primitive_date_time(block_data.block.timestamp());
                 // TODO: store blocks
                 // TODO: remove substates (I think). These can be requested lazily and cached (LRU) as needed to allow
                 // TODO: an committed transaction should queue a shard state sync in the affected shards
                 // an upper bound on substates stored in the indexer.
-                self.store_substates_in_db(&block_data.diff, timestamp)?;
+                info!(
+                    target: LOG_TARGET,
+                    "Storing {} substate update(s) for block {} (epoch={}, height={})",
+                    block_data.diff.len(),
+                    block_data.block.id(),
+                    block_data.block.epoch(),
+                    block_data.block.height()
+                );
+                self.store_substates_in_db(&block_data.diff)?;
             }
         }
 
@@ -125,42 +127,24 @@ impl BlockScanner {
             .map_err(|e| e.into())
     }
 
-    fn store_substates_in_db(
-        &self,
-        updates: &[SubstateUpdateProof],
-        timestamp: PrimitiveDateTime,
-    ) -> Result<(), anyhow::Error> {
+    fn store_substates_in_db(&self, updates: &[SubstateUpdateProof]) -> Result<(), anyhow::Error> {
         let mut tx = self.substate_store.create_write_tx()?;
         // store/update up substates if any
         for update in updates {
             match update {
                 SubstateUpdateProof::Create(create) => {
-                    let maybe_substate_value = create.substate.value.value();
-                    if maybe_substate_value.is_none() {
+                    if create.substate.value.value().is_none() {
                         warn!(
                             target: LOG_TARGET,
                             "⚠️ Received UP substate {} without value. This indicates that the substate has been pruned. Some event data is not available.", create.substate.as_versioned_substate_id_ref(),
                         );
                     }
-                    let template_address = maybe_substate_value.and_then(Self::extract_template_address_from_substate);
-                    let module_name = maybe_substate_value.and_then(Self::extract_module_name_from_substate);
-                    let substate_row = NewSubstate {
-                        address: create.substate.substate_id.to_string(),
-                        version: create.substate.version as i32,
-                        data: maybe_substate_value
-                            .map(Self::encode_substate)
-                            .transpose()?
-                            .unwrap_or_default(),
-                        template_address: template_address.map(|s| s.to_string()),
-                        module_name,
-                        timestamp,
-                    };
                     debug!(
                         target: LOG_TARGET,
                         "Saving substate: {:?}",
-                        substate_row
+                        create.substate
                     );
-                    tx.upsert_substate(substate_row)?;
+                    tx.upsert_substate(&create.substate)?;
                 },
                 SubstateUpdateProof::Destroy(_) => {},
             }
@@ -169,25 +153,6 @@ impl BlockScanner {
         Ok(())
     }
 
-    fn extract_template_address_from_substate(substate: &SubstateValue) -> Option<TemplateAddress> {
-        match substate {
-            SubstateValue::Component(c) => Some(c.template_address),
-            _ => None,
-        }
-    }
-
-    fn extract_module_name_from_substate(substate: &SubstateValue) -> Option<String> {
-        match substate {
-            SubstateValue::Component(c) => Some(c.module_name.to_owned()),
-            _ => None,
-        }
-    }
-
-    fn encode_substate(substate: &SubstateValue) -> Result<String, anyhow::Error> {
-        let pretty_json = serde_json::to_string_pretty(&substate)?;
-        Ok(pretty_json)
-    }
-
     async fn get_oldest_scanned_epoch(&self) -> Result<Option<Epoch>, anyhow::Error> {
         self.substate_store
             .with_read_tx(|tx| tx.get_oldest_scanned_epoch())
@@ -361,29 +326,3 @@ impl BlockScanner {
         Ok(blocks)
     }
 }
-
-fn unix_epoch_to_primitive_date_time(timestamp: u64) -> PrimitiveDateTime {
-    let timestamp = i64::try_from(timestamp).unwrap_or_else(|e| {
-        // TODO: this is very possible because we trust that the timestamp is roughly correct, however
-        // it is purely informational and not enforced in consensus therefore could be any value and
-        // therefore cannot be relied for ordering (use (epoch,height) instead).
-        warn!(
-            target: LOG_TARGET,
-            "Failed to convert block timestamp to PrimitiveDateTime: {}",
-            e
-        );
-        i64::MAX // = August 17, 292278994, 07:12:55.807 UTC
-    });
-    OffsetDateTime::from_unix_timestamp(timestamp)
-        .map(|osdt| PrimitiveDateTime::new(osdt.date(), osdt.time()))
-        .unwrap_or_else(|e| {
-            warn!(
-                target: LOG_TARGET,
-                "Failed to convert block timestamp to OffsetDateTime: {}. Using UNIX_EPOCH",
-                e
-            );
-            // An error cannot be because the timestamp is too small, because we use an u64 and a zero unix
-            // timestamp represents a greater date (1970 AD) than the minimum (9999 BC)
-            PrimitiveDateTime::MAX
-        })
-}

applications/tari_indexer/src/network_state_sync/worker.rs

@@ -22,7 +22,7 @@ use tari_ootle_common_types::{
 };
 use tari_ootle_p2p::{proto::rpc, TariMessagingSpec};
 use tari_ootle_storage::{
-    consensus_models::{EpochCheckpoint, SubstateUpdateProof, SubstateValueFilterFlags},
+    consensus_models::{EpochCheckpoint, SubstateData, SubstateUpdateProof, SubstateValueFilterFlags},
     StorageError,
 };
 use tari_rpc_framework::__macro_reexports::future::Either;
@@ -282,6 +282,7 @@ impl NetworkWideStateSync {
         let mut update_buf = Vec::new();
         let mut utxos_buf = Vec::new();
         let mut transactions_buf = Vec::new();
+        let mut validator_fee_pools_buf = Vec::new();
 
         let mut has_synced_global_shard = false;
 
@@ -297,6 +298,7 @@ impl NetworkWideStateSync {
                     &mut update_buf,
                     &mut utxos_buf,
                     &mut transactions_buf,
+                    &mut validator_fee_pools_buf,
                     shard_group,
                     &mut session,
                 )
@@ -311,6 +313,7 @@ impl NetworkWideStateSync {
                     &mut update_buf,
                     &mut utxos_buf,
                     &mut transactions_buf,
+                    &mut validator_fee_pools_buf,
                     shard_group,
                     &mut session,
                 )
@@ -328,6 +331,7 @@ impl NetworkWideStateSync {
         update_buf: &mut Vec<(Epoch, SubstateUpdateProof)>,
         utxos_buf: &mut Vec<UtxoUpdateRecord>,
         transactions_buf: &mut Vec<TransactionReceipt>,
+        validator_fee_pools_buf: &mut Vec<SubstateData>,
         shard_group: ShardGroup,
         session: &mut ValidatorRpcSession,
     ) -> Result<(), NetworkStateSyncError> {
@@ -348,6 +352,7 @@ impl NetworkWideStateSync {
                 until_epoch: None,
                 value_filters: (SubstateValueFilterFlags::UTXO |
                     SubstateValueFilterFlags::TEMPLATE |
+                    SubstateValueFilterFlags::VALIDATOR_FEE_POOL |
                     SubstateValueFilterFlags::TRANSACTION_RECEIPT)
                     .bits(),
             })
@@ -389,6 +394,7 @@ impl NetworkWideStateSync {
                     &mut templates_buf,
                     utxos_buf,
                     transactions_buf,
+                    validator_fee_pools_buf,
                 )?;
             }
             if msg.has_more {
@@ -405,6 +411,11 @@ impl NetworkWideStateSync {
                 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(..))?;
+                // 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(..) {
+                    tx.upsert_substate(&substate_data)?;
+                }
                 // TODO: transaction events and templates
                 debug!(target: LOG_TARGET, "✅ Committing {} transactions for shard {shard} (epoch: {msg_epoch})", transactions_buf.len());
                 self.stats.increase_events(transactions_buf.len());
@@ -453,6 +464,7 @@ fn extend_bufs_from_substate_update(
     templates_buf: &mut Vec<TemplateChange>,
     utxos_buf: &mut Vec<UtxoUpdateRecord>,
     transactions_buf: &mut Vec<TransactionReceipt>,
+    validator_fee_pools_buf: &mut Vec<SubstateData>,
 ) -> Result<(), NetworkStateSyncError> {
     match &update {
         SubstateUpdateProof::Create(create) => match create.substate.value().value() {
@@ -489,6 +501,13 @@ fn extend_bufs_from_substate_update(
                     warn!(target: LOG_TARGET, "⚠️ NEVER HAPPEN: Received template substate with invalid address: {}", create.substate.substate_id());
                 }
             },
+            Some(SubstateValue::ValidatorFeePool(_)) => {
+                validator_fee_pools_buf.push(SubstateData {
+                    substate_id: create.substate.substate_id().clone(),
+                    version: create.substate.version,
+                    value: create.substate.value().clone(),
+                });
+            },
             Some(_) => {},
             None => {
                 let id = create.substate.substate_id();

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

@@ -6,8 +6,6 @@ create table substates
     data             text      not NULL,
     template_address text      NULL,
     module_name      text      NULL,
-    -- Block timestamp
-    timestamp        timestamp not NULL,
     updated_at       timestamp not null default current_timestamp,
     created_at       timestamp not null default current_timestamp
 );

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

@@ -43,7 +43,6 @@ pub struct SubstateRecord {
     pub data: String,
     pub template_address: Option<String>,
     pub module_name: Option<String>,
-    pub timestamp: PrimitiveDateTime,
     pub updated_at: PrimitiveDateTime,
     pub created_at: PrimitiveDateTime,
 }
@@ -72,5 +71,4 @@ pub struct NewSubstate {
     pub data: String,
     pub template_address: Option<String>,
     pub module_name: Option<String>,
-    pub timestamp: PrimitiveDateTime,
 }

applications/tari_indexer/src/storage_sqlite/reader.rs

@@ -100,7 +100,7 @@ impl IndexerStoreReadTransaction for SqliteStoreReadTransaction<'_> {
                 let substate_id = SubstateId::from_str(&s.address)?;
                 let version = s.version as u32;
                 let template_address = s.template_address.map(|h| deserialize_hex_try_from(&h)).transpose()?;
-                let timestamp = s.timestamp;
+                let timestamp = s.updated_at;
                 Ok(ListSubstateItem {
                     substate_id,
                     module_name: s.module_name,

applications/tari_indexer/src/storage_sqlite/schema.rs

@@ -74,7 +74,6 @@ diesel::table! {
         data -> Text,
         template_address -> Nullable<Text>,
         module_name -> Nullable<Text>,
-        timestamp -> Timestamp,
         updated_at -> Timestamp,
         created_at -> Timestamp,
     }

applications/tari_indexer/src/storage_sqlite/store_factory.rs

@@ -17,7 +17,7 @@ use tari_engine_types::{events::Event, substate::SubstateId, Utxo};
 use tari_indexer_client::types::{ListSubstateItem, NonFungibleSubstate, TransactionEntry};
 use tari_ootle_common_types::{shard::Shard, substate_type::SubstateType, Epoch, ShardGroup, StateVersion};
 use tari_ootle_storage::{
-    consensus_models::{EpochCheckpoint, SubstateUpdateProof},
+    consensus_models::{EpochCheckpoint, SubstateData, SubstateUpdateProof},
     StorageError,
 };
 use tari_ootle_storage_sqlite::{error::SqliteStorageError, SqliteTransaction};
@@ -33,7 +33,7 @@ use tari_transaction::{Transaction, TransactionId};
 
 use crate::{
     storage_sqlite::{
-        models::{EventRecord, KeyValue, NewScannedBlockId, NewSubstate, SubstateRecord, UtxoUpdateRecord},
+        models::{EventRecord, KeyValue, NewScannedBlockId, SubstateRecord, UtxoUpdateRecord},
         reader::SqliteStoreReadTransaction,
         writer::SqliteStoreWriteTransaction,
     },
@@ -209,7 +209,7 @@ pub trait IndexerStoreWriteTransaction {
         &mut self,
         updates: I,
     ) -> Result<(), StorageError>;
-    fn upsert_substate(&mut self, new_substate: NewSubstate) -> Result<(), StorageError>;
+    fn upsert_substate(&mut self, substate: &SubstateData) -> Result<(), StorageError>;
     fn batch_insert_events<I: IntoIterator<Item = Event>>(&mut self, events: I) -> Result<(), StorageError>;
     fn save_scanned_block_id(&mut self, new_scanned_block_id: NewScannedBlockId) -> Result<(), StorageError>;
     fn delete_scanned_epochs_older_than(&mut self, epoch: Epoch) -> Result<(), StorageError>;

applications/tari_indexer/src/storage_sqlite/writer.rs

@@ -9,7 +9,7 @@ use serde::Serialize;
 use tari_engine_types::events::Event;
 use tari_ootle_common_types::{shard::Shard, substate_type::SubstateType, Epoch, StateVersion};
 use tari_ootle_storage::{
-    consensus_models::{EpochCheckpoint, SubstateUpdateProof},
+    consensus_models::{EpochCheckpoint, SubstateData, SubstateUpdateProof},
     StorageError,
 };
 use tari_ootle_storage_sqlite::SqliteTransaction;
@@ -170,9 +170,32 @@ impl IndexerStoreWriteTransaction for SqliteStoreWriteTransaction<'_> {
         Ok(())
     }
 
-    fn upsert_substate(&mut self, new_substate: NewSubstate) -> Result<(), StorageError> {
+    fn upsert_substate(&mut self, substate: &SubstateData) -> Result<(), StorageError> {
         use crate::storage_sqlite::schema::substates;
 
+        let template_address = substate
+            .value
+            .value()
+            .and_then(|s| s.component())
+            .map(|c| c.template_address.to_string());
+        let module_name = substate
+            .value
+            .value()
+            .and_then(|s| s.component())
+            .map(|c| c.module_name.clone());
+        let new_substate = NewSubstate {
+            address: substate.substate_id.to_string(),
+            version: substate.version as i32,
+            data: substate
+                .value
+                .value()
+                .map(serialize_json)
+                .transpose()?
+                .unwrap_or_default(),
+            template_address,
+            module_name,
+        };
+
         let address = &new_substate.address;
         let current_substate = substates::table
             .filter(substates::address.eq(address))

applications/tari_indexer/web_ui/package.json

@@ -7,7 +7,7 @@
     "dev": "vite",
     "build": "tsc && vite build",
     "preview": "vite preview",
-    "clean-dist": "rm -rf dist && rm -f tsconfig.tsbuildinfo"
+    "clean-dist": "rm -rf dist/* && rm -f tsconfig.tsbuildinfo"
   },
   "dependencies": {
     "@emotion/react": "^11.14.0",

applications/tari_swarm_daemon/src/process_definitions/wallet_daemon_create_key.rs

@@ -29,19 +29,7 @@ impl ProcessDefinition for WalletDaemonCreateAccount {
             .arg("-b")
             .arg(context.base_path())
             .arg("--network")
-            .arg(context.network().to_string())
-            .args([
-                "create-account",
-                "--name",
-                "Fees",
-                "--key",
-                "0",
-                "--set-active",
-                "--output",
-                output_path
-                    .to_str()
-                    .context("Non-UTF8 output path in WalletDaemonCreateAccount")?,
-            ]);
+            .arg(context.network().to_string());
 
         if let Some(override_keyring_password) =
             context.get_setting(wallet_daemon::OVERRIDE_KEYRING_PASSWORD_SETTINGS_KEY)
@@ -51,6 +39,19 @@ impl ProcessDefinition for WalletDaemonCreateAccount {
                 .arg(override_keyring_password);
         }
 
+        command.args([
+            "create-account",
+            "--name",
+            "Validator Fees",
+            "--key",
+            "0",
+            "--set-active",
+            "--output",
+            output_path
+                .to_str()
+                .context("Non-UTF8 output path in WalletDaemonCreateAccount")?,
+        ]);
+
         Ok(command)
     }