Add detailed sync progress

Author: buffrr

client/src/bin/spaced.rs

@@ -8,7 +8,7 @@ use spaces_client::{
     rpc::{AsyncChainState, RpcServerImpl, WalletLoadRequest, WalletManager},
     source::{BitcoinBlockSource, BitcoinRpc},
     store,
-    sync::Spaced,
+    spaces::Spaced,
     wallets::RpcWallet,
 };
 use store::LiveSnapshot;

client/src/cbf.rs

@@ -13,6 +13,7 @@ use spaces_wallet::bitcoin::ScriptBuf;
 use spaces_wallet::SpacesWallet;
 use crate::client::{BlockSource, BlockchainInfo};
 use crate::source::BitcoinBlockSource;
+use crate::wallets::WalletProgressUpdate;
 
 pub struct CompactFilterSync {
     graph: IndexedTxGraph<ConfirmationBlockTime, KeychainTxOutIndex<KeychainKind>>,
@@ -24,6 +25,8 @@ pub struct CompactFilterSync {
     queued_blocks: BTreeMap<u32, BlockHash>,
     queued_filters: VecDeque<u32>,
     filters_tip: u32,
+    block_matches: u32,
+    total_filters: u32,
     wait: Option<Instant>,
     state: SyncState,
 }
@@ -36,7 +39,7 @@ enum SyncState {
     WaitForBlocks,
     ProcessBlocks,
     ApplyUpdate,
-    Synced
+    Synced,
 }
 
 impl CompactFilterSync {
@@ -56,6 +59,8 @@ impl CompactFilterSync {
             queued_blocks: BTreeMap::new(),
             queued_filters: Default::default(),
             filters_tip: 0,
+            block_matches: 0,
+            total_filters: 0,
             wait: None,
             state: SyncState::SyncChecks,
         };
@@ -101,6 +106,7 @@ impl CompactFilterSync {
         &mut self,
         wallet: &mut SpacesWallet,
         source: &BitcoinBlockSource,
+        progress: &mut WalletProgressUpdate,
     ) -> anyhow::Result<()> {
         if self.wait.is_some_and(|w| w.elapsed() < Duration::from_secs(10)) {
             return Ok(());
@@ -112,11 +118,16 @@ impl CompactFilterSync {
                 let info = source.get_blockchain_info()?;
                 if info.headers != info.blocks {
                     info!("Source still syncing, retrying...");
+                    *progress = WalletProgressUpdate::Syncing;
                     self.wait = Some(Instant::now());
                     return Ok(());
                 }
                 if info.filters != info.filter_headers {
                     info!("Filters syncing, retrying...");
+                    *progress = WalletProgressUpdate::CbfFilterSync {
+                        total: info.filter_headers.unwrap_or(0),
+                        completed: info.filters.unwrap_or(0),
+                    };
                     self.wait = Some(Instant::now());
                     return Ok(());
                 }
@@ -156,14 +167,15 @@ impl CompactFilterSync {
                     self.queued_filters.push_back(height);
                 }
                 self.filters_tip = end;
+                self.total_filters = self.queued_filters.len() as u32;
                 self.state = SyncState::ProcessFilters;
             }
             SyncState::ProcessFilters => {
                 let height = match self.queued_filters.pop_front() {
                     None => {
                         self.state = SyncState::QueueBlocks;
-                        return Ok(())
-                    },
+                        return Ok(());
+                    }
                     Some(f) => f,
                 };
                 let idx_filter = source.get_block_filter_by_height(height)?;
@@ -173,10 +185,15 @@ impl CompactFilterSync {
                 if filter.match_any(&idx_filter.hash, self.scripts.iter().map(|s| s.as_bytes()))? {
                     self.queued_blocks.insert(height, idx_filter.hash);
                     self.load_more_scripts(wallet);
+                    self.block_matches += 1;
                     info!("wallet({}) processed block filter {} - match found", wallet.name(), height);
                 } else {
                     info!("wallet({}) processed block filter {} - no match", wallet.name(), height);
                 }
+                *progress = WalletProgressUpdate::CbfProcessFilters {
+                    total: self.total_filters,
+                    completed: self.total_filters - self.queued_filters.len() as u32,
+                };
             }
             SyncState::QueueBlocks => {
                 if !self.queued_blocks.is_empty() {
@@ -195,6 +212,16 @@ impl CompactFilterSync {
 
                 if status.pending > 0 {
                     info!("wallet({}): waiting for {} pending blocks", wallet.name(), status.pending);
+
+                    // The client has a global state for pending blocks in the queue
+                    // so we cap it just in case other things are queuing blocks
+                    // at the same time
+                    let pending = std::cmp::min(status.pending, self.block_matches);
+                    *progress = WalletProgressUpdate::CbfDownloadMatchingBlocks {
+                        total: self.block_matches,
+                        completed: self.block_matches - pending,
+                    };
+
                     self.wait = Some(Instant::now());
                     return Ok(());
                 }
@@ -209,7 +236,8 @@ impl CompactFilterSync {
             SyncState::ProcessBlocks => {
                 let (height, hash) = match self.queued_blocks.pop_first() {
                     None => {
-                        self.state = SyncState::Synced;
+                        *progress = WalletProgressUpdate::CbfApplyUpdate;
+                        self.state = SyncState::ApplyUpdate;
                         return Ok(());
                     }
                     Some(f) => f,
@@ -219,7 +247,10 @@ impl CompactFilterSync {
                     .ok_or(anyhow!("block {} {} not found", height, hash))?;
                 self.chain_changeset.insert(height, Some(hash));
                 let _ = self.graph.apply_block_relevant(&block, height);
-                self.state = SyncState::ApplyUpdate;
+                *progress = WalletProgressUpdate::CbfProcessMatchingBlocks {
+                    total: self.block_matches,
+                    completed: self.block_matches - self.queued_blocks.len() as u32 ,
+                };
             }
             SyncState::ApplyUpdate => {
                 info!("wallet({}): updating wallet tip to {}", wallet.name(), self.filters_tip);
@@ -233,8 +264,10 @@ impl CompactFilterSync {
                 wallet.insert_checkpoint(filters_anchor)?;
                 info!("wallet({}): compact filter sync portion complete at {}", wallet.name(), self.filters_tip);
                 self.state = SyncState::Synced;
+                // Only CBF portion is done
+                *progress = WalletProgressUpdate::Syncing
             }
-            SyncState::Synced => {}
+            SyncState::Synced => {},
         }
         Ok(())
     }

client/src/config.rs

@@ -22,7 +22,7 @@ use toml::Value;
 use crate::{
     source::{BitcoinRpc, BitcoinRpcAuth},
     store::{LiveStore, Store},
-    sync::Spaced,
+    spaces::Spaced,
 };
 
 const RPC_OPTIONS: &str = "RPC Server Options";

client/src/format.rs

@@ -14,14 +14,15 @@ use spaces_wallet::{
         BidEventDetails, BidoutEventDetails, OpenEventDetails, SendEventDetails,
         TransferEventDetails, TxEventKind,
     },
-    Balance, DoubleUtxo, WalletInfo, WalletOutput,
+    Balance, DoubleUtxo, WalletOutput,
 };
 use tabled::{Table, Tabled};
 
 use crate::{
     rpc::ServerInfo,
     wallets::{ListSpacesResponse, TxInfo, TxResponse, WalletResponse},
 };
+use crate::wallets::{WalletInfoWithProgress, WalletProgressUpdate};
 
 #[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Debug, ValueEnum, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
@@ -163,27 +164,62 @@ pub fn print_server_info(info: ServerInfo, format: Format) {
     }
 }
 
-pub fn print_wallet_info(info: WalletInfo, format: Format) {
+pub fn print_wallet_info(prog: WalletInfoWithProgress, format: Format) {
     match format {
         Format::Text => {
-            println!("WALLET: {}", info.label);
-            println!("  Tip {}\n  Birthday {}", info.tip, info.start_block);
+            println!("WALLET: {}", prog.info.label);
+            println!("  Tip {}\n  Birthday {}", prog.info.tip, prog.info.start_block);
 
             println!("  Public descriptors");
-            for desc in info.descriptors {
+            for desc in prog.info.descriptors {
                 println!("    {}", desc.descriptor);
             }
+
+            // Print sync status
+            println!("  Sync Status:");
+            match prog.status {
+                WalletProgressUpdate::SourceSync { total, completed } => {
+                    println!("    Source Syncing: {}/{} ({:.1}%)", completed, total,
+                             (completed as f64 / total as f64) * 100.0);
+                }
+                WalletProgressUpdate::CbfFilterSync { total, completed } => {
+                    println!("    Filters Syncing: {}/{} ({:.1}%)", completed, total,
+                             (completed as f64 / total as f64) * 100.0);
+                }
+                WalletProgressUpdate::CbfProcessFilters { total, completed } => {
+                    println!("    Processing Filters: {}/{} ({:.1}%)", completed, total,
+                             (completed as f64 / total as f64) * 100.0);
+                }
+                WalletProgressUpdate::CbfDownloadMatchingBlocks { total, completed } => {
+                    println!("    Downloading Matching Blocks: {}/{} ({:.1}%)", completed, total,
+                             (completed as f64 / total as f64) * 100.0);
+                }
+                WalletProgressUpdate::CbfProcessMatchingBlocks { total, completed } => {
+                    println!("    Processing Matching Blocks: {}/{} ({:.1}%)", completed, total,
+                             (completed as f64 / total as f64) * 100.0);
+                }
+                WalletProgressUpdate::Syncing => {
+                    println!("    Syncing: In progress ({:.1}%):", prog.info.progress * 100.0);
+                }
+                WalletProgressUpdate::CbfApplyUpdate => {
+                    println!("    Applying compact filters update");
+                }
+                WalletProgressUpdate::Complete => {
+                    println!("    Complete");
+                }
+            }
+
             println!();
         }
         Format::Json => {
-            println!("{}", serde_json::to_string_pretty(&info).unwrap());
+            println!("{}", serde_json::to_string_pretty(&prog.info).unwrap());
         }
     }
 }
 
 fn ascii_table<I, T>(iter: I) -> String
 where
-    I: IntoIterator<Item = T>,
+    I: IntoIterator<Item=T>,
     T: Tabled,
 {
     Table::new(iter)

client/src/lib.rs

@@ -16,7 +16,7 @@ pub mod format;
 pub mod rpc;
 pub mod source;
 pub mod store;
-pub mod sync;
+pub mod spaces;
 pub mod wallets;
 mod cbf;
 
@@ -60,3 +60,15 @@ where
         Vec::<u8>::deserialize(deserializer)
     }
 }
+
+pub fn calc_progress(start_block: u32, tip: u32, chain_tip: u32) -> f32 {
+    if chain_tip <= start_block || tip < start_block {
+        0.0
+    } else if tip >= chain_tip {
+        1.0
+    } else {
+        let blocks_synced = tip - start_block;
+        let blocks_to_sync = chain_tip - start_block;
+        blocks_synced as f32 / blocks_to_sync as f32
+    }
+}

client/src/rpc.rs

@@ -40,27 +40,19 @@ use spaces_protocol::{
 use spaces_wallet::{
     bdk_wallet as bdk, bdk_wallet::template::Bip86, bitcoin::hashes::Hash as BitcoinHash,
     export::WalletExport, nostr::NostrEvent, Balance, DoubleUtxo, Listing, SpacesWallet,
-    WalletConfig, WalletDescriptors, WalletInfo, WalletOutput,
+    WalletConfig, WalletDescriptors, WalletOutput,
 };
 use tokio::{
     select,
     sync::{broadcast, mpsc, oneshot, RwLock},
     task::JoinSet,
 };
 
-use crate::{
-    checker::TxChecker,
-    client::{BlockMeta, TxEntry},
-    config::ExtendedNetwork,
-    deserialize_base64, serialize_base64,
-    source::BitcoinRpc,
-    store::{ChainState, LiveSnapshot, RolloutEntry, Sha256},
-    sync::{COMMIT_BLOCK_INTERVAL, ROOT_ANCHORS_COUNT},
-    wallets::{
-        AddressKind, ListSpacesResponse, RpcWallet, TxInfo, TxResponse, WalletCommand,
-        WalletResponse,
-    },
-};
+use crate::{calc_progress, checker::TxChecker, client::{BlockMeta, TxEntry}, config::ExtendedNetwork, deserialize_base64, serialize_base64, source::BitcoinRpc, store::{ChainState, LiveSnapshot, RolloutEntry, Sha256}, spaces::{COMMIT_BLOCK_INTERVAL, ROOT_ANCHORS_COUNT}, wallets::{
+    AddressKind, ListSpacesResponse, RpcWallet, TxInfo, TxResponse, WalletCommand,
+    WalletResponse,
+}};
+use crate::wallets::WalletInfoWithProgress;
 
 pub(crate) type Responder<T> = oneshot::Sender<T>;
 
@@ -229,7 +221,7 @@ pub trait Rpc {
     ) -> Result<NostrEvent, ErrorObjectOwned>;
 
     #[method(name = "walletgetinfo")]
-    async fn wallet_get_info(&self, name: &str) -> Result<WalletInfo, ErrorObjectOwned>;
+    async fn wallet_get_info(&self, name: &str) -> Result<WalletInfoWithProgress, ErrorObjectOwned>;
 
     #[method(name = "walletexport")]
     async fn wallet_export(&self, name: &str) -> Result<WalletExport, ErrorObjectOwned>;
@@ -870,7 +862,7 @@ impl RpcServer for RpcServerImpl {
             .map_err(|error| ErrorObjectOwned::owned(-1, error.to_string(), None::<String>))
     }
 
-    async fn wallet_get_info(&self, wallet: &str) -> Result<WalletInfo, ErrorObjectOwned> {
+    async fn wallet_get_info(&self, wallet: &str) -> Result<WalletInfoWithProgress, ErrorObjectOwned> {
         self.wallet(&wallet)
             .await?
             .send_get_info()
@@ -1584,17 +1576,21 @@ async fn get_server_info(client: &reqwest::Client, rpc: &BitcoinRpc, tip: ChainA
         .await
         .map_err(|e| anyhow!("Could not retrieve blockchain info ({})", e))?;
 
+    let start_block = if info.chain == "main" {
+        871_222
+    } else if info.chain.starts_with("test") {
+        50_000
+    } else {
+        0
+    };
+
     Ok(ServerInfo {
         network: info.chain,
         tip,
         chain: ChainInfo {
             blocks: info.blocks,
             headers: info.headers,
         },
-        progress: if info.headers != 0 && info.headers >= tip.height {
-            tip.height as f32 / info.headers as f32
-        } else {
-            0.0
-        },
+        progress: calc_progress(start_block, tip.height, info.headers),
     })
 }