Add UTXO resync notification feature

Implemented the UTXO resync notification feature in the UTXO Index, notifying the user when the UTXO index starts resyncing. This is done by adding a new UtxoResyncState in the SyncStateChangedNotification enum and notifying the user when the resync event occurs. Dependencies were updated in Cargo.toml and the notification was integrated into the GRPC layer.

consensus/notify/src/notification.rs

@@ -180,6 +180,7 @@ pub enum SyncStateChangedNotification {
     Proof { current: u8, max: u8 }, // todo other variants
     Headers { headers: u64, progress: i64 },
     Blocks { blocks: u64, progress: i64 },
+    UtxoResync,
 }
 
 impl SyncStateChangedNotification {
@@ -194,4 +195,8 @@ impl SyncStateChangedNotification {
     pub fn new_blocks(blocks: u64, progress: i64) -> SyncStateChangedNotification {
         Self::Blocks { blocks, progress }
     }
+
+    pub fn new_resync_utxo() -> SyncStateChangedNotification {
+        Self::UtxoResync
+    }
 }

indexes/utxoindex/Cargo.toml

@@ -9,13 +9,16 @@ license.workspace = true
 
 [dependencies]
 kaspa-hashes.workspace = true
-thiserror.workspace = true
 kaspa-consensus-core.workspace = true
+kaspa-consensusmanager.workspace = true
+kaspa-consensus-notify.workspace = true
+kaspa-notify.workspace = true
 kaspa-core.workspace = true
 kaspa-utils.workspace = true
 kaspa-index-core.workspace = true
 kaspa-database.workspace = true
-kaspa-consensusmanager.workspace = true
+
+thiserror.workspace = true
 log.workspace = true
 rocksdb.workspace = true
 serde.workspace = true

indexes/utxoindex/src/index.rs

@@ -7,6 +7,14 @@ use crate::{
     IDENT,
 };
 use kaspa_consensus_core::{tx::ScriptPublicKeys, utxo::utxo_diff::UtxoDiff, BlockHashSet};
+
+#[cfg(not(test))]
+use kaspa_consensus_notify::notification::{Notification, SyncStateChangedNotification};
+#[cfg(not(test))]
+use kaspa_consensus_notify::root::ConsensusNotificationRoot;
+#[cfg(not(test))]
+use kaspa_notify::notifier::Notify;
+
 use kaspa_consensusmanager::{ConsensusManager, ConsensusResetHandler};
 use kaspa_core::{info, trace};
 use kaspa_database::prelude::{StoreError, StoreResult, DB};
@@ -26,12 +34,23 @@ const RESYNC_CHUNK_SIZE: usize = 2048; //Increased from 1k (used in go-kaspad),
 pub struct UtxoIndex {
     consensus_manager: Arc<ConsensusManager>,
     store: Store,
+    #[cfg(not(test))]
+    notification_root: Arc<ConsensusNotificationRoot>,
 }
 
 impl UtxoIndex {
     /// Creates a new [`UtxoIndex`] within a [`RwLock`]
-    pub fn new(consensus_manager: Arc<ConsensusManager>, db: Arc<DB>) -> UtxoIndexResult<Arc<RwLock<Self>>> {
-        let mut utxoindex = Self { consensus_manager: consensus_manager.clone(), store: Store::new(db) };
+    pub fn new(
+        consensus_manager: Arc<ConsensusManager>,
+        db: Arc<DB>,
+        #[cfg(not(test))] notification_root: Arc<ConsensusNotificationRoot>,
+    ) -> UtxoIndexResult<Arc<RwLock<Self>>> {
+        let mut utxoindex = Self {
+            consensus_manager: consensus_manager.clone(),
+            store: Store::new(db),
+            #[cfg(not(test))]
+            notification_root,
+        };
         if !utxoindex.is_synced()? {
             utxoindex.resync()?;
         }
@@ -128,6 +147,11 @@ impl UtxoIndexApi for UtxoIndex {
     fn resync(&mut self) -> UtxoIndexResult<()> {
         info!("Resyncing the utxoindex...");
 
+        #[cfg(not(test))]
+        self.notification_root
+            .notify(Notification::SyncStateChanged(SyncStateChangedNotification::new_resync_utxo()))
+            .expect("expecting an open unbounded channel");
+
         self.store.delete_all()?;
         let consensus = self.consensus_manager.consensus();
         let session = futures::executor::block_on(consensus.session_blocking());

rpc/core/src/convert/notification.rs

@@ -104,6 +104,7 @@ impl From<&consensus_notify::SyncStateChangedNotification> for SyncStateChangedN
             consensus_notify::SyncStateChangedNotification::Blocks { blocks, progress } => {
                 Self::Blocks { blocks: *blocks, progress: *progress }
             }
+            consensus_notify::SyncStateChangedNotification::UtxoResync => Self::UtxoResync,
         }
     }
 }

rpc/core/src/model/message.rs

@@ -1090,6 +1090,7 @@ pub enum SyncStateChangedNotification {
         #[allow(dead_code)]
         progress: i64,
     },
+    UtxoResync,
 }
 
 ///

rpc/grpc/core/proto/rpc.proto

@@ -777,6 +777,7 @@ message SyncStateChangedNotificationMessage {
     ProofState proof = 1;
     HeadersState headers = 2;
     BlocksState blocks = 3;
+    UtxoResyncState utxoResync = 4;
   }
 }
 
@@ -793,4 +794,6 @@ message HeadersState {
 message BlocksState {
   uint64 blocks = 1;
   int64 progress = 2;
-}
+}
+
+message UtxoResyncState {}

rpc/grpc/core/src/convert/notification.rs

@@ -94,6 +94,9 @@ from!(item: &kaspa_rpc_core::SyncStateChangedNotification, SyncStateChangedNotif
                     progress: *progress
                 }))
             },
+            SyncStateChangedNotification::UtxoResync => SyncStateChangedNotificationMessage {
+                sync_state: Some(sync_state_changed_notification_message::SyncState::UtxoResync(crate::protowire::UtxoResyncState{}))
+            },
         }
 });