Skip to content

Commit 8723bf9

Browse files
pauldeluciapauldelucia
authored
feat: spv progress bars and sync from genesis when wallets are loaded (#454)
* feat: spv progress bars * feat: start from 0 if syncing with wallets * clippy
1 parent bba90a4 commit 8723bf9

File tree

2 files changed

+338
-130
lines changed

2 files changed

+338
-130
lines changed

‎src/spv/manager.rs‎

Lines changed: 128 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,9 @@ use crate::config::NetworkConfig;
33
use crate::utils::tasks::TaskManager;
44
use dash_sdk::dash_spv::network::MultiPeerNetworkManager;
55
use dash_sdk::dash_spv::storage::DiskStorageManager;
6-
use dash_sdk::dash_spv::types::{DetailedSyncProgress, SpvEvent, SyncProgress, ValidationMode};
6+
use dash_sdk::dash_spv::types::{
7+
DetailedSyncProgress, SpvEvent, SyncProgress, SyncStage, ValidationMode,
8+
};
79
use dash_sdk::dash_spv::{ClientConfig, DashSpvClient};
810
use dash_sdk::dpp::dashcore::Network;
911
use dash_sdk::dpp::key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
@@ -103,6 +105,9 @@ pub struct SpvManager {
103105
status: Arc<RwLock<SpvStatus>>,
104106
last_error: Arc<RwLock<Option<String>>>,
105107
started_at: Arc<RwLock<Option<SystemTime>>>,
108+
sync_progress_state: Arc<RwLock<Option<SyncProgress>>>,
109+
detailed_progress_state: Arc<RwLock<Option<DetailedSyncProgress>>>,
110+
progress_updated_at: Arc<RwLock<Option<SystemTime>>>,
106111
// mapping DET wallet seed_hash -> SPV wallet identifier (if created)
107112
det_wallets: Arc<RwLock<std::collections::BTreeMap<[u8; 32], WalletId>>>,
108113
// signal channel to trigger external reconcile on wallet-related events
@@ -132,6 +137,9 @@ impl SpvManager {
132137
status: Arc::new(RwLock::new(SpvStatus::Idle)),
133138
last_error: Arc::new(RwLock::new(None)),
134139
started_at: Arc::new(RwLock::new(None)),
140+
sync_progress_state: Arc::new(RwLock::new(None)),
141+
detailed_progress_state: Arc::new(RwLock::new(None)),
142+
progress_updated_at: Arc::new(RwLock::new(None)),
135143
det_wallets: Arc::new(RwLock::new(std::collections::BTreeMap::new())),
136144
reconcile_tx: Mutex::new(None),
137145
stop_token: Mutex::new(None),
@@ -142,7 +150,7 @@ impl SpvManager {
142150

143151
/// Async status method for getting full details including progress
144152
pub async fn status_async(&self) -> SpvStatusSnapshot {
145-
let client_guard = self.client.read().await;
153+
let _client_guard = self.client.read().await;
146154
let status = *self.status.read().expect("SPV status lock poisoned");
147155
let last_error = self
148156
.last_error
@@ -153,23 +161,29 @@ impl SpvManager {
153161
.started_at
154162
.read()
155163
.expect("SPV started_at lock poisoned");
156-
157-
// Get progress directly from the client if available
158-
let (sync_progress, detailed_progress) = if let Some(_client) = client_guard.as_ref() {
159-
// Note: These would need to be exposed by dash-spv's DashSpvClient
160-
// For now, we'll track them separately until dash-spv exposes them
161-
(None, None)
162-
} else {
163-
(None, None)
164-
};
164+
let sync_progress = self
165+
.sync_progress_state
166+
.read()
167+
.expect("SPV sync_progress lock poisoned")
168+
.clone();
169+
let detailed_progress = self
170+
.detailed_progress_state
171+
.read()
172+
.expect("SPV detailed_progress lock poisoned")
173+
.clone();
174+
let last_updated = (*self
175+
.progress_updated_at
176+
.read()
177+
.expect("SPV progress_updated lock poisoned"))
178+
.or(Some(SystemTime::now()));
165179

166180
SpvStatusSnapshot {
167181
status,
168182
sync_progress,
169183
detailed_progress,
170184
last_error,
171185
started_at,
172-
last_updated: Some(SystemTime::now()),
186+
last_updated,
173187
}
174188
}
175189

@@ -185,14 +199,29 @@ impl SpvManager {
185199
.started_at
186200
.read()
187201
.expect("SPV started_at lock poisoned");
202+
let sync_progress = self
203+
.sync_progress_state
204+
.read()
205+
.expect("SPV sync_progress lock poisoned")
206+
.clone();
207+
let detailed_progress = self
208+
.detailed_progress_state
209+
.read()
210+
.expect("SPV detailed_progress lock poisoned")
211+
.clone();
212+
let last_updated = (*self
213+
.progress_updated_at
214+
.read()
215+
.expect("SPV progress_updated lock poisoned"))
216+
.or(Some(SystemTime::now()));
188217

189218
SpvStatusSnapshot {
190219
status,
191-
sync_progress: None,
192-
detailed_progress: None,
220+
sync_progress,
221+
detailed_progress,
193222
last_error,
194223
started_at,
195-
last_updated: Some(SystemTime::now()),
224+
last_updated,
196225
}
197226
}
198227

@@ -217,6 +246,18 @@ impl SpvManager {
217246
.started_at
218247
.write()
219248
.expect("SPV started_at lock poisoned") = Some(SystemTime::now());
249+
*self
250+
.sync_progress_state
251+
.write()
252+
.expect("SPV sync_progress lock poisoned") = None;
253+
*self
254+
.detailed_progress_state
255+
.write()
256+
.expect("SPV detailed_progress lock poisoned") = None;
257+
*self
258+
.progress_updated_at
259+
.write()
260+
.expect("SPV progress_updated lock poisoned") = None;
220261

221262
let stop_token = CancellationToken::new();
222263
*self
@@ -485,9 +526,24 @@ impl SpvManager {
485526
// Sync to tip
486527
match client.sync_to_tip().await {
487528
Ok(progress) => {
488-
tracing::info!("Initial sync complete: {:?}", progress);
529+
tracing::info!("Initial sync progress snapshot: {:?}", progress);
530+
{
531+
let mut stored_sync = self
532+
.sync_progress_state
533+
.write()
534+
.expect("SPV sync_progress lock poisoned");
535+
*stored_sync = Some(progress.clone());
536+
}
537+
{
538+
let mut updated_at = self
539+
.progress_updated_at
540+
.write()
541+
.expect("SPV progress_updated lock poisoned");
542+
*updated_at = Some(SystemTime::now());
543+
}
544+
// Stay in Syncing mode until detailed progress reports completion.
489545
*self.status.write().expect("SPV status lock poisoned") =
490-
SpvStatus::Running;
546+
SpvStatus::Syncing;
491547
}
492548
Err(err) => {
493549
tracing::error!("Initial sync failed: {}", err);
@@ -557,6 +613,10 @@ impl SpvManager {
557613
mut progress_rx: tokio::sync::mpsc::UnboundedReceiver<DetailedSyncProgress>,
558614
) {
559615
let status = Arc::clone(&self.status);
616+
let last_error = Arc::clone(&self.last_error);
617+
let sync_progress_state = Arc::clone(&self.sync_progress_state);
618+
let detailed_progress_state = Arc::clone(&self.detailed_progress_state);
619+
let progress_updated_at = Arc::clone(&self.progress_updated_at);
560620
let cancel = self.subtasks.cancellation_token.clone();
561621

562622
self.subtasks.spawn_sync(async move {
@@ -569,14 +629,49 @@ impl SpvManager {
569629
msg = progress_rx.recv() => {
570630
match msg {
571631
Some(detailed) => {
632+
{
633+
let mut stored_detailed = detailed_progress_state
634+
.write()
635+
.expect("SPV detailed_progress lock poisoned");
636+
*stored_detailed = Some(detailed.clone());
637+
}
638+
{
639+
let mut stored_sync = sync_progress_state
640+
.write()
641+
.expect("SPV sync_progress lock poisoned");
642+
*stored_sync = Some(detailed.sync_progress.clone());
643+
}
644+
{
645+
let mut updated_at = progress_updated_at
646+
.write()
647+
.expect("SPV progress_updated lock poisoned");
648+
*updated_at = Some(detailed.last_update_time);
649+
}
650+
572651
if last_update.elapsed() >= min_interval {
573-
// Update status based on progress
574-
if detailed.percentage >= 100.0 || detailed.sync_progress.header_height >= detailed.peer_best_height {
575-
*status.write().expect("SPV status lock poisoned") = SpvStatus::Running;
576-
} else {
577-
let current = *status.read().expect("SPV status lock poisoned");
578-
if matches!(current, SpvStatus::Starting | SpvStatus::Idle | SpvStatus::Stopped) {
579-
*status.write().expect("SPV status lock poisoned") = SpvStatus::Syncing;
652+
// Update status based on progress stage and completeness
653+
let mut status_guard = status
654+
.write()
655+
.expect("SPV status lock poisoned");
656+
let current = *status_guard;
657+
match &detailed.sync_stage {
658+
SyncStage::Complete => {
659+
*status_guard = SpvStatus::Running;
660+
}
661+
SyncStage::Failed(message) => {
662+
*status_guard = SpvStatus::Error;
663+
let mut err_guard = last_error
664+
.write()
665+
.expect("SPV last_error lock poisoned");
666+
*err_guard = Some(format!("SPV sync failed: {message}"));
667+
}
668+
_ => {
669+
if !matches!(
670+
current,
671+
SpvStatus::Stopping | SpvStatus::Stopped | SpvStatus::Error
672+
) {
673+
*status_guard = SpvStatus::Syncing;
674+
}
580675
}
581676
}
582677
last_update = std::time::Instant::now();
@@ -634,12 +729,18 @@ impl SpvManager {
634729
>,
635730
String,
636731
> {
732+
let start_height = {
733+
let guard = self.wallet.read().await;
734+
if guard.wallet_count() == 0 {
735+
u32::MAX
736+
} else {
737+
0
738+
}
739+
};
637740
let mut config = ClientConfig::new(self.network)
638741
.with_storage_path(self.data_dir.clone())
639742
.with_validation_mode(ValidationMode::Full)
640-
// Start from the latest built-in checkpoint instead of genesis
641-
// (effective only when storage is empty / first initialization)
642-
.with_start_height(u32::MAX);
743+
.with_start_height(start_height);
643744

644745
// Pin peers when running against local nodes to avoid random peers.
645746
if self.network == Network::Devnet || self.network == Network::Regtest {

0 commit comments

Comments
 (0)