1use std::collections::{BTreeMap, HashMap};
4use std::ops::Range;
5use std::sync::Arc;
6use std::sync::atomic::{self, AtomicBool, AtomicU8};
7use std::time::{Duration, SystemTime};
8
9use tokio::sync::{RwLock, mpsc};
10
11use incrementalmerkletree::{Marking, Retention};
12use orchard::tree::MerkleHashOrchard;
13use shardtree::store::ShardStore;
14use zcash_keys::keys::UnifiedFullViewingKey;
15use zcash_primitives::transaction::{Transaction, TxId};
16use zcash_protocol::ShieldedProtocol;
17use zcash_protocol::consensus::{self, BlockHeight};
18use zingo_netutils::lightwallet_protocol::RawTransaction;
19use zingo_netutils::{Indexer, TransparentIndexer};
20use zip32::AccountId;
21
22use zingo_status::confirmation_status::ConfirmationStatus;
23
24use crate::client::{self, FetchRequest};
25use crate::config::{PerformanceLevel, SyncConfig};
26use crate::error::{
27 ContinuityError, MempoolError, ScanError, ServerError, SyncError, SyncModeError,
28 SyncStatusError,
29};
30use crate::keys::transparent::TransparentAddressId;
31use crate::scan::ScanResults;
32use crate::scan::task::{Scanner, ScannerState};
33use crate::scan::transactions::scan_transaction;
34use crate::sync::state::truncate_scan_ranges;
35use crate::wallet::traits::{
36 SyncBlocks, SyncNullifiers, SyncOutPoints, SyncShardTrees, SyncTransactions, SyncWallet,
37};
38use crate::wallet::{
39 KeyIdInterface, NoteInterface, NullifierMap, OutputId, OutputInterface, ScanTarget, SyncMode,
40 SyncState, WalletBlock, WalletTransaction,
41};
42use crate::witness::LocatedTreeData;
43
44#[cfg(not(feature = "darkside_test"))]
45use crate::witness;
46
47#[cfg(not(feature = "darkside_test"))]
48pub(crate) mod transparent;
49
50pub(crate) mod spend;
51pub(crate) mod state;
52
53pub(crate) const MAX_REORG_ALLOWANCE: u32 = 100;
54const VERIFY_BLOCK_RANGE_SIZE: u32 = 10;
55
56#[derive(Debug, Clone)]
61#[allow(missing_docs)]
62pub struct SyncStatus {
63 pub scan_ranges: Vec<ScanRange>,
64 pub sync_start_height: BlockHeight,
65 pub session_blocks_scanned: u32,
66 pub total_blocks_scanned: u32,
67 pub percentage_session_blocks_scanned: f32,
68 pub percentage_total_blocks_scanned: f32,
69 pub session_sapling_outputs_scanned: u32,
70 pub total_sapling_outputs_scanned: u32,
71 pub session_orchard_outputs_scanned: u32,
72 pub total_orchard_outputs_scanned: u32,
73 pub percentage_session_outputs_scanned: f32,
74 pub percentage_total_outputs_scanned: f32,
75}
76
77impl std::fmt::Display for SyncStatus {
79 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80 write!(
81 f,
82 "percentage complete: {}",
83 self.percentage_total_outputs_scanned
84 )
85 }
86}
87
88impl From<SyncStatus> for json::JsonValue {
89 fn from(value: SyncStatus) -> Self {
90 let scan_ranges: Vec<json::JsonValue> = value
91 .scan_ranges
92 .iter()
93 .map(|range| {
94 json::object! {
95 "priority" => format!("{:?}", range.priority()),
96 "start_block" => range.block_range().start.to_string(),
97 "end_block" => (range.block_range().end - 1).to_string(),
98 }
99 })
100 .collect();
101
102 json::object! {
103 "scan_ranges" => scan_ranges,
104 "sync_start_height" => u32::from(value.sync_start_height),
105 "session_blocks_scanned" => value.session_blocks_scanned,
106 "total_blocks_scanned" => value.total_blocks_scanned,
107 "percentage_session_blocks_scanned" => value.percentage_session_blocks_scanned,
108 "percentage_total_blocks_scanned" => value.percentage_total_blocks_scanned,
109 "session_sapling_outputs_scanned" => value.session_sapling_outputs_scanned,
110 "total_sapling_outputs_scanned" => value.total_sapling_outputs_scanned,
111 "session_orchard_outputs_scanned" => value.session_orchard_outputs_scanned,
112 "total_orchard_outputs_scanned" => value.total_orchard_outputs_scanned,
113 "percentage_session_outputs_scanned" => value.percentage_session_outputs_scanned,
114 "percentage_total_outputs_scanned" => value.percentage_total_outputs_scanned,
115 }
116 }
117}
118
119#[derive(Debug, Clone)]
121#[allow(missing_docs)]
122pub struct SyncResult {
123 pub sync_start_height: BlockHeight,
124 pub sync_end_height: BlockHeight,
125 pub blocks_scanned: u32,
126 pub sapling_outputs_scanned: u32,
127 pub orchard_outputs_scanned: u32,
128 pub percentage_total_outputs_scanned: f32,
129}
130
131impl std::fmt::Display for SyncResult {
132 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133 write!(
134 f,
135 "Sync completed succesfully:
136{{
137 sync start height: {}
138 sync end height: {}
139 blocks scanned: {}
140 sapling outputs scanned: {}
141 orchard outputs scanned: {}
142 percentage total outputs scanned: {}
143}}",
144 self.sync_start_height,
145 self.sync_end_height,
146 self.blocks_scanned,
147 self.sapling_outputs_scanned,
148 self.orchard_outputs_scanned,
149 self.percentage_total_outputs_scanned,
150 )
151 }
152}
153
154impl From<SyncResult> for json::JsonValue {
155 fn from(value: SyncResult) -> Self {
156 json::object! {
157 "sync_start_height" => u32::from(value.sync_start_height),
158 "sync_end_height" => u32::from(value.sync_end_height),
159 "blocks_scanned" => value.blocks_scanned,
160 "sapling_outputs_scanned" => value.sapling_outputs_scanned,
161 "orchard_outputs_scanned" => value.orchard_outputs_scanned,
162 "percentage_total_outputs_scanned" => value.percentage_total_outputs_scanned,
163 }
164 }
165}
166
167#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
169pub enum ScanPriority {
170 RefetchingNullifiers,
172 Scanning,
174 Scanned,
176 ScannedWithoutMapping,
181 Historic,
183 OpenAdjacent,
185 FoundNote,
187 ChainTip,
189 Verify,
192}
193
194#[derive(Debug, Clone, PartialEq, Eq)]
196pub struct ScanRange {
197 block_range: Range<BlockHeight>,
198 priority: ScanPriority,
199}
200
201impl std::fmt::Display for ScanRange {
202 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
203 write!(
204 f,
205 "{:?}({}..{})",
206 self.priority, self.block_range.start, self.block_range.end,
207 )
208 }
209}
210
211impl ScanRange {
212 #[must_use]
214 pub fn from_parts(block_range: Range<BlockHeight>, priority: ScanPriority) -> Self {
215 assert!(
216 block_range.end >= block_range.start,
217 "{block_range:?} is invalid for ScanRange({priority:?})",
218 );
219 ScanRange {
220 block_range,
221 priority,
222 }
223 }
224
225 #[must_use]
227 pub fn block_range(&self) -> &Range<BlockHeight> {
228 &self.block_range
229 }
230
231 #[must_use]
233 pub fn priority(&self) -> ScanPriority {
234 self.priority
235 }
236
237 #[must_use]
239 pub fn is_empty(&self) -> bool {
240 self.block_range.is_empty()
241 }
242
243 #[must_use]
245 pub fn len(&self) -> usize {
246 usize::try_from(u32::from(self.block_range.end) - u32::from(self.block_range.start))
247 .expect("due to number of max blocks should always be valid usize")
248 }
249
250 #[must_use]
254 pub fn truncate_start(&self, block_height: BlockHeight) -> Option<Self> {
255 if block_height >= self.block_range.end || self.is_empty() {
256 None
257 } else {
258 Some(ScanRange {
259 block_range: self.block_range.start.max(block_height)..self.block_range.end,
260 priority: self.priority,
261 })
262 }
263 }
264
265 #[must_use]
269 pub fn truncate_end(&self, block_height: BlockHeight) -> Option<Self> {
270 if block_height <= self.block_range.start || self.is_empty() {
271 None
272 } else {
273 Some(ScanRange {
274 block_range: self.block_range.start..self.block_range.end.min(block_height),
275 priority: self.priority,
276 })
277 }
278 }
279
280 #[must_use]
284 pub fn split_at(&self, p: BlockHeight) -> Option<(Self, Self)> {
285 (p > self.block_range.start && p < self.block_range.end).then_some((
286 ScanRange {
287 block_range: self.block_range.start..p,
288 priority: self.priority,
289 },
290 ScanRange {
291 block_range: p..self.block_range.end,
292 priority: self.priority,
293 },
294 ))
295 }
296}
297
298pub async fn sync<C, P, W>(
309 client: C,
310 consensus_parameters: &P,
311 wallet: Arc<RwLock<W>>,
312 sync_mode: Arc<AtomicU8>,
313 config: SyncConfig,
314) -> Result<SyncResult, SyncError<W::Error>>
315where
316 C: Clone + Indexer + TransparentIndexer + Sync + Send + 'static,
317 P: consensus::Parameters + Sync + Send + 'static,
318 W: SyncWallet
319 + SyncBlocks
320 + SyncTransactions
321 + SyncNullifiers
322 + SyncOutPoints
323 + SyncShardTrees
324 + Send,
325{
326 let mut sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?;
327 if sync_mode_enum == SyncMode::NotRunning {
328 sync_mode_enum = SyncMode::Running;
329 sync_mode.store(sync_mode_enum as u8, atomic::Ordering::Release);
330 } else {
331 return Err(SyncModeError::SyncAlreadyRunning.into());
332 }
333
334 tracing::info!("Starting sync...");
335
336 let (fetch_request_sender, fetch_request_receiver) = mpsc::unbounded_channel();
338 let client_clone = client.clone();
339 let fetcher_handle =
340 tokio::spawn(
341 async move { client::fetch::fetch(fetch_request_receiver, client_clone).await },
342 );
343
344 let (mempool_transaction_sender, mut mempool_transaction_receiver) = mpsc::channel(100);
346 let shutdown_mempool = Arc::new(AtomicBool::new(false));
347 let shutdown_mempool_clone = shutdown_mempool.clone();
348 let unprocessed_mempool_transactions_count = Arc::new(AtomicU8::new(0));
349 let unprocessed_mempool_transactions_count_clone =
350 unprocessed_mempool_transactions_count.clone();
351 let mempool_handle = tokio::spawn(async move {
352 mempool_monitor(
353 client,
354 mempool_transaction_sender,
355 unprocessed_mempool_transactions_count_clone,
356 shutdown_mempool_clone,
357 )
358 .await
359 });
360
361 let mut wallet_guard = wallet.write().await;
363
364 let chain_height = client::get_chain_height(fetch_request_sender.clone()).await?;
365 if chain_height == 0.into() {
366 return Err(SyncError::ServerError(ServerError::GenesisBlockOnly));
367 }
368 let last_known_chain_height =
369 checked_wallet_height(&mut *wallet_guard, chain_height, consensus_parameters)?;
370
371 let ufvks = wallet_guard
372 .get_unified_full_viewing_keys()
373 .map_err(SyncError::WalletError)?;
374
375 #[cfg(not(feature = "darkside_test"))]
376 transparent::update_addresses_and_scan_targets(
377 consensus_parameters,
378 &mut *wallet_guard,
379 fetch_request_sender.clone(),
380 &ufvks,
381 last_known_chain_height,
382 chain_height,
383 config.transparent_address_discovery,
384 )
385 .await?;
386
387 #[cfg(not(feature = "darkside_test"))]
388 update_subtree_roots(
389 consensus_parameters,
390 fetch_request_sender.clone(),
391 &mut *wallet_guard,
392 )
393 .await?;
394
395 add_initial_frontier(
396 consensus_parameters,
397 fetch_request_sender.clone(),
398 &mut *wallet_guard,
399 )
400 .await?;
401
402 let initial_reorg_detection_start_height = state::update_scan_ranges(
403 consensus_parameters,
404 fetch_request_sender.clone(),
405 last_known_chain_height,
406 chain_height,
407 &mut *wallet_guard,
408 )
409 .await?;
410
411 state::set_initial_state(
412 consensus_parameters,
413 fetch_request_sender.clone(),
414 &mut *wallet_guard,
415 chain_height,
416 )
417 .await?;
418
419 expire_transactions(&mut *wallet_guard)?;
420
421 drop(wallet_guard);
422
423 let (scan_results_sender, mut scan_results_receiver) = mpsc::unbounded_channel();
425 let mut scanner = Scanner::new(
426 consensus_parameters.clone(),
427 scan_results_sender,
428 fetch_request_sender.clone(),
429 ufvks.clone(),
430 );
431 scanner.launch(config.performance_level);
432
433 let mut nullifier_map_limit_exceeded = false;
436 let mut interval = tokio::time::interval(Duration::from_millis(50));
437 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
438 loop {
439 tokio::select! {
440 Some((scan_range, scan_results)) = scan_results_receiver.recv() => {
441 let mut wallet_guard = wallet.write().await;
442 process_scan_results(
443 consensus_parameters,
444 &mut *wallet_guard,
445 fetch_request_sender.clone(),
446 &ufvks,
447 scan_range,
448 scan_results,
449 initial_reorg_detection_start_height,
450 config.performance_level,
451 &mut nullifier_map_limit_exceeded,
452 )
453 .await?;
454 wallet_guard.set_save_flag().map_err(SyncError::WalletError)?;
455 drop(wallet_guard);
456 }
457
458 Some(raw_transaction) = mempool_transaction_receiver.recv() => {
459 let mut wallet_guard = wallet.write().await;
460 process_mempool_transaction(
461 consensus_parameters,
462 &ufvks,
463 &mut *wallet_guard,
464 raw_transaction,
465 )
466 .await?;
467 unprocessed_mempool_transactions_count.fetch_sub(1, atomic::Ordering::Release);
468 drop(wallet_guard);
469 }
470
471 _update_scanner = interval.tick() => {
472 sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?;
473 match sync_mode_enum {
474 SyncMode::Paused => {
475 let mut pause_interval = tokio::time::interval(Duration::from_secs(1));
476 pause_interval.tick().await;
477 while sync_mode_enum == SyncMode::Paused {
478 pause_interval.tick().await;
479 sync_mode_enum = SyncMode::from_atomic_u8(sync_mode.clone())?;
480 }
481 },
482 SyncMode::Shutdown => {
483 let mut wallet_guard = wallet.write().await;
484 let sync_status = match sync_status(&*wallet_guard).await {
485 Ok(status) => status,
486 Err(SyncStatusError::WalletError(e)) => {
487 return Err(SyncError::WalletError(e));
488 }
489 Err(SyncStatusError::NoSyncData) => {
490 panic!("sync data must exist!");
491 }
492 };
493 wallet_guard
494 .set_save_flag()
495 .map_err(SyncError::WalletError)?;
496 drop(wallet_guard);
497 mempool_handle.abort();
498 fetcher_handle.abort();
499 tracing::info!("Sync successfully shutdown.");
500
501 return Ok(SyncResult {
502 sync_start_height: sync_status.sync_start_height,
503 sync_end_height: (sync_status
504 .scan_ranges
505 .last()
506 .expect("should be non-empty after syncing")
507 .block_range()
508 .end
509 - 1),
510 blocks_scanned: sync_status.session_blocks_scanned,
511 sapling_outputs_scanned: sync_status.session_sapling_outputs_scanned,
512 orchard_outputs_scanned: sync_status.session_orchard_outputs_scanned,
513 percentage_total_outputs_scanned: sync_status.percentage_total_outputs_scanned,
514 });
515 }
516 SyncMode::Running => (),
517 SyncMode::NotRunning => {
518 panic!("sync mode should not be manually set to NotRunning!");
519 },
520 }
521
522 scanner.update(&mut *wallet.write().await, shutdown_mempool.clone(), nullifier_map_limit_exceeded).await?;
523
524 if matches!(scanner.state, ScannerState::Shutdown) {
525 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
527 if is_shutdown(&scanner, unprocessed_mempool_transactions_count.clone())
528 {
529 tracing::info!("Sync successfully shutdown.");
530 break;
531 }
532 }
533 }
534 }
535 }
536
537 let mut wallet_guard = wallet.write().await;
538 let sync_status = match sync_status(&*wallet_guard).await {
539 Ok(status) => status,
540 Err(SyncStatusError::WalletError(e)) => {
541 return Err(SyncError::WalletError(e));
542 }
543 Err(SyncStatusError::NoSyncData) => {
544 panic!("sync data must exist!");
545 }
546 };
547 for transaction in wallet_guard
549 .get_wallet_transactions_mut()
550 .map_err(SyncError::WalletError)?
551 .values_mut()
552 {
553 for note in transaction.sapling_notes.as_mut_slice() {
554 note.refetch_nullifier_ranges = Vec::new();
555 }
556 for note in transaction.orchard_notes.as_mut_slice() {
557 note.refetch_nullifier_ranges = Vec::new();
558 }
559 }
560 wallet_guard
561 .set_save_flag()
562 .map_err(SyncError::WalletError)?;
563
564 drop(wallet_guard);
565 drop(scanner);
566 drop(fetch_request_sender);
567
568 match mempool_handle.await.expect("task panicked") {
569 Ok(()) => (),
570 Err(e @ MempoolError::ShutdownWithoutStream) => tracing::warn!("{e}"),
571 Err(e) => return Err(e.into()),
572 }
573 fetcher_handle.await.expect("task panicked");
574
575 Ok(SyncResult {
576 sync_start_height: sync_status.sync_start_height,
577 sync_end_height: (sync_status
578 .scan_ranges
579 .last()
580 .expect("should be non-empty after syncing")
581 .block_range()
582 .end
583 - 1),
584 blocks_scanned: sync_status.session_blocks_scanned,
585 sapling_outputs_scanned: sync_status.session_sapling_outputs_scanned,
586 orchard_outputs_scanned: sync_status.session_orchard_outputs_scanned,
587 percentage_total_outputs_scanned: sync_status.percentage_total_outputs_scanned,
588 })
589}
590
591fn checked_wallet_height<W, P>(
607 wallet: &mut W,
608 chain_height: BlockHeight,
609 consensus_parameters: &P,
610) -> Result<BlockHeight, SyncError<W::Error>>
611where
612 W: SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees,
613 P: zcash_protocol::consensus::Parameters,
614{
615 let sync_state = wallet.get_sync_state().map_err(SyncError::WalletError)?;
616 if let Some(last_known_chain_height) = sync_state.last_known_chain_height() {
617 if last_known_chain_height > chain_height {
618 if last_known_chain_height - chain_height >= MAX_REORG_ALLOWANCE {
619 return Err(SyncError::ChainError(
623 u32::from(last_known_chain_height),
624 MAX_REORG_ALLOWANCE,
625 u32::from(chain_height),
626 ));
627 }
628 truncate_wallet_data(wallet, chain_height)?;
631 truncate_scan_ranges(
632 chain_height,
633 wallet
634 .get_sync_state_mut()
635 .map_err(SyncError::WalletError)?,
636 );
637 return Ok(chain_height);
638 }
639 Ok(last_known_chain_height)
641 } else {
642 let sapling_activation_height = consensus_parameters
644 .activation_height(consensus::NetworkUpgrade::Sapling)
645 .expect("sapling activation height should always return Some");
646 let birthday = wallet.get_birthday().map_err(SyncError::WalletError)?;
647 if birthday > chain_height {
648 return Err(SyncError::ChainError(
651 u32::from(birthday),
652 MAX_REORG_ALLOWANCE,
653 u32::from(chain_height),
654 ));
655 } else if birthday < sapling_activation_height {
656 return Err(SyncError::BirthdayBelowSapling(
657 u32::from(birthday),
658 u32::from(sapling_activation_height),
659 ));
660 }
661
662 Ok(birthday - 1)
663 }
664}
665
666pub async fn sync_status<W>(wallet: &W) -> Result<SyncStatus, SyncStatusError<W::Error>>
672where
673 W: SyncWallet + SyncBlocks,
674{
675 let (total_sapling_outputs_scanned, total_orchard_outputs_scanned) =
676 state::calculate_scanned_outputs(wallet).map_err(SyncStatusError::WalletError)?;
677 let total_outputs_scanned = total_sapling_outputs_scanned + total_orchard_outputs_scanned;
678
679 let sync_state = wallet
680 .get_sync_state()
681 .map_err(SyncStatusError::WalletError)?;
682 if sync_state.initial_sync_state.sync_start_height == 0.into() {
683 return Ok(SyncStatus {
684 scan_ranges: sync_state.scan_ranges.clone(),
685 sync_start_height: 0.into(),
686 session_blocks_scanned: 0,
687 total_blocks_scanned: 0,
688 percentage_session_blocks_scanned: 0.0,
689 percentage_total_blocks_scanned: 0.0,
690 session_sapling_outputs_scanned: 0,
691 session_orchard_outputs_scanned: 0,
692 total_sapling_outputs_scanned: 0,
693 total_orchard_outputs_scanned: 0,
694 percentage_session_outputs_scanned: 0.0,
695 percentage_total_outputs_scanned: 0.0,
696 });
697 }
698 let total_blocks_scanned = state::calculate_scanned_blocks(sync_state);
699
700 let birthday = sync_state
701 .wallet_birthday()
702 .ok_or(SyncStatusError::NoSyncData)?;
703 let last_known_chain_height = sync_state
704 .last_known_chain_height()
705 .ok_or(SyncStatusError::NoSyncData)?;
706 let total_blocks = last_known_chain_height - birthday + 1;
707 let total_sapling_outputs = sync_state
708 .initial_sync_state
709 .wallet_tree_bounds
710 .sapling_final_tree_size
711 - sync_state
712 .initial_sync_state
713 .wallet_tree_bounds
714 .sapling_initial_tree_size;
715 let total_orchard_outputs = sync_state
716 .initial_sync_state
717 .wallet_tree_bounds
718 .orchard_final_tree_size
719 - sync_state
720 .initial_sync_state
721 .wallet_tree_bounds
722 .orchard_initial_tree_size;
723 let total_outputs = total_sapling_outputs + total_orchard_outputs;
724
725 let session_blocks_scanned =
726 total_blocks_scanned - sync_state.initial_sync_state.previously_scanned_blocks;
727 let mut percentage_session_blocks_scanned = ((session_blocks_scanned as f32
728 / (total_blocks - sync_state.initial_sync_state.previously_scanned_blocks) as f32)
729 * 100.0)
730 .clamp(0.0, 100.0);
731 let mut percentage_total_blocks_scanned =
732 ((total_blocks_scanned as f32 / total_blocks as f32) * 100.0).clamp(0.0, 100.0);
733
734 let session_sapling_outputs_scanned = total_sapling_outputs_scanned
735 - sync_state
736 .initial_sync_state
737 .previously_scanned_sapling_outputs;
738 let session_orchard_outputs_scanned = total_orchard_outputs_scanned
739 - sync_state
740 .initial_sync_state
741 .previously_scanned_orchard_outputs;
742 let session_outputs_scanned = session_sapling_outputs_scanned + session_orchard_outputs_scanned;
743 let previously_scanned_outputs = sync_state
744 .initial_sync_state
745 .previously_scanned_sapling_outputs
746 + sync_state
747 .initial_sync_state
748 .previously_scanned_orchard_outputs;
749 let mut percentage_session_outputs_scanned = ((session_outputs_scanned as f32
750 / (total_outputs - previously_scanned_outputs) as f32)
751 * 100.0)
752 .clamp(0.0, 100.0);
753 let mut percentage_total_outputs_scanned =
754 ((total_outputs_scanned as f32 / total_outputs as f32) * 100.0).clamp(0.0, 100.0);
755
756 if sync_state.scan_ranges().iter().any(|scan_range| {
757 scan_range.priority() == ScanPriority::ScannedWithoutMapping
758 || scan_range.priority() == ScanPriority::RefetchingNullifiers
759 }) {
760 if percentage_session_blocks_scanned == 100.0 {
761 percentage_session_blocks_scanned = 99.0;
762 }
763 if percentage_total_blocks_scanned == 100.0 {
764 percentage_total_blocks_scanned = 99.0;
765 }
766 if percentage_session_outputs_scanned == 100.0 {
767 percentage_session_outputs_scanned = 99.0;
768 }
769 if percentage_total_outputs_scanned == 100.0 {
770 percentage_total_outputs_scanned = 99.0;
771 }
772 }
773
774 Ok(SyncStatus {
775 scan_ranges: sync_state.scan_ranges.clone(),
776 sync_start_height: sync_state.initial_sync_state.sync_start_height,
777 session_blocks_scanned,
778 total_blocks_scanned,
779 percentage_session_blocks_scanned,
780 percentage_total_blocks_scanned,
781 session_sapling_outputs_scanned,
782 total_sapling_outputs_scanned,
783 session_orchard_outputs_scanned,
784 total_orchard_outputs_scanned,
785 percentage_session_outputs_scanned,
786 percentage_total_outputs_scanned,
787 })
788}
789
790pub fn scan_pending_transaction<W>(
797 consensus_parameters: &impl consensus::Parameters,
798 ufvks: &HashMap<AccountId, UnifiedFullViewingKey>,
799 wallet: &mut W,
800 transaction: Transaction,
801 status: ConfirmationStatus,
802 datetime: u32,
803) -> Result<(), SyncError<W::Error>>
804where
805 W: SyncWallet + SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees,
806{
807 if matches!(status, ConfirmationStatus::Confirmed(_)) {
808 panic!("this fn is for unconfirmed transactions only");
809 }
810
811 let mut pending_transaction_nullifiers = NullifierMap::new();
812 let mut pending_transaction_outpoints = BTreeMap::new();
813 let transparent_addresses: HashMap<String, TransparentAddressId> = wallet
814 .get_transparent_addresses()
815 .map_err(SyncError::WalletError)?
816 .iter()
817 .map(|(id, address)| (address.clone(), *id))
818 .collect();
819 let pending_transaction = scan_transaction(
820 consensus_parameters,
821 ufvks,
822 transaction.txid(),
823 transaction,
824 status,
825 None,
826 &mut pending_transaction_nullifiers,
827 &mut pending_transaction_outpoints,
828 &transparent_addresses,
829 datetime,
830 )?;
831
832 let wallet_transactions = wallet
833 .get_wallet_transactions()
834 .map_err(SyncError::WalletError)?;
835 let transparent_output_ids = spend::collect_transparent_output_ids(wallet_transactions);
836 let transparent_spend_scan_targets = spend::detect_transparent_spends(
837 &mut pending_transaction_outpoints,
838 transparent_output_ids,
839 );
840 let (sapling_derived_nullifiers, orchard_derived_nullifiers) =
841 spend::collect_derived_nullifiers(wallet_transactions);
842 let (sapling_spend_scan_targets, orchard_spend_scan_targets) = spend::detect_shielded_spends(
843 &mut pending_transaction_nullifiers,
844 sapling_derived_nullifiers,
845 orchard_derived_nullifiers,
846 );
847
848 if pending_transaction.transparent_coins().is_empty()
850 && pending_transaction.sapling_notes().is_empty()
851 && pending_transaction.orchard_notes().is_empty()
852 && pending_transaction.outgoing_orchard_notes().is_empty()
853 && pending_transaction.outgoing_sapling_notes().is_empty()
854 && transparent_spend_scan_targets.is_empty()
855 && sapling_spend_scan_targets.is_empty()
856 && orchard_spend_scan_targets.is_empty()
857 {
858 return Ok(());
859 }
860
861 wallet
862 .insert_wallet_transaction(pending_transaction)
863 .map_err(SyncError::WalletError)?;
864 spend::update_spent_coins(
865 wallet
866 .get_wallet_transactions_mut()
867 .map_err(SyncError::WalletError)?,
868 transparent_spend_scan_targets,
869 );
870 spend::update_spent_notes(
871 wallet,
872 sapling_spend_scan_targets,
873 orchard_spend_scan_targets,
874 false,
875 )
876 .map_err(SyncError::WalletError)?;
877
878 Ok(())
879}
880
881pub fn add_scan_targets(sync_state: &mut SyncState, scan_targets: &[ScanTarget]) {
892 for scan_target in scan_targets {
893 sync_state.scan_targets.insert(*scan_target);
894 }
895}
896
897pub fn reset_spends(
903 wallet_transactions: &mut HashMap<TxId, WalletTransaction>,
904 invalid_txids: Vec<TxId>,
905) {
906 wallet_transactions
907 .values_mut()
908 .flat_map(|transaction| transaction.orchard_notes_mut())
909 .filter(|output| {
910 output
911 .spending_transaction
912 .is_some_and(|spending_txid| invalid_txids.contains(&spending_txid))
913 })
914 .for_each(|output| {
915 output.set_spending_transaction(None);
916 });
917 wallet_transactions
918 .values_mut()
919 .flat_map(|transaction| transaction.sapling_notes_mut())
920 .filter(|output| {
921 output
922 .spending_transaction
923 .is_some_and(|spending_txid| invalid_txids.contains(&spending_txid))
924 })
925 .for_each(|output| {
926 output.set_spending_transaction(None);
927 });
928 wallet_transactions
929 .values_mut()
930 .flat_map(|transaction| transaction.transparent_coins_mut())
931 .filter(|output| {
932 output
933 .spending_transaction
934 .is_some_and(|spending_txid| invalid_txids.contains(&spending_txid))
935 })
936 .for_each(|output| {
937 output.set_spending_transaction(None);
938 });
939}
940
941pub fn set_transactions_failed(
945 wallet_transactions: &mut HashMap<TxId, WalletTransaction>,
946 failed_txids: Vec<TxId>,
947) {
948 for failed_txid in failed_txids.iter() {
949 if let Some(transaction) = wallet_transactions.get_mut(failed_txid) {
950 let height = transaction.status().get_height();
951 transaction.update_status(
952 ConfirmationStatus::Failed(height),
953 SystemTime::now()
954 .duration_since(SystemTime::UNIX_EPOCH)
955 .expect("infalliable for such long time periods")
956 .as_secs() as u32,
957 );
958 }
959 }
960 reset_spends(wallet_transactions, failed_txids);
961}
962
963fn is_shutdown<P>(
965 scanner: &Scanner<P>,
966 mempool_unprocessed_transactions_count: Arc<AtomicU8>,
967) -> bool
968where
969 P: consensus::Parameters + Sync + Send + 'static,
970{
971 scanner.worker_poolsize() == 0
972 && mempool_unprocessed_transactions_count.load(atomic::Ordering::Acquire) == 0
973}
974
975#[allow(clippy::too_many_arguments)]
977async fn process_scan_results<W>(
978 consensus_parameters: &impl consensus::Parameters,
979 wallet: &mut W,
980 fetch_request_sender: mpsc::UnboundedSender<FetchRequest>,
981 ufvks: &HashMap<AccountId, UnifiedFullViewingKey>,
982 scan_range: ScanRange,
983 scan_results: Result<ScanResults, ScanError>,
984 initial_reorg_detection_start_height: BlockHeight,
985 performance_level: PerformanceLevel,
986 nullifier_map_limit_exceeded: &mut bool,
987) -> Result<(), SyncError<W::Error>>
988where
989 W: SyncWallet
990 + SyncBlocks
991 + SyncTransactions
992 + SyncNullifiers
993 + SyncOutPoints
994 + SyncShardTrees
995 + Send,
996{
997 match scan_results {
998 Ok(results) => {
999 let ScanResults {
1000 mut nullifiers,
1001 mut outpoints,
1002 scanned_blocks,
1003 wallet_transactions,
1004 sapling_located_trees,
1005 orchard_located_trees,
1006 } = results;
1007
1008 if scan_range.priority() == ScanPriority::ScannedWithoutMapping {
1009 let full_refetching_nullifiers_range = wallet
1012 .get_sync_state()
1013 .map_err(SyncError::WalletError)?
1014 .scan_ranges
1015 .iter()
1016 .find(|&wallet_scan_range| {
1017 wallet_scan_range
1018 .block_range()
1019 .contains(&scan_range.block_range().start)
1020 && wallet_scan_range
1021 .block_range()
1022 .contains(&(scan_range.block_range().end - 1))
1023 })
1024 .expect("wallet scan range containing scan range should exist!");
1025 if scan_range.block_range().start
1026 != full_refetching_nullifiers_range.block_range().start
1027 || scan_range.block_range().end
1028 != full_refetching_nullifiers_range.block_range().end
1029 {
1030 let mut missing_block_bounds = BTreeMap::new();
1031 for block_bound in [
1032 scan_range.block_range().start - 1,
1033 scan_range.block_range().start,
1034 scan_range.block_range().end - 1,
1035 scan_range.block_range().end,
1036 ] {
1037 if block_bound < full_refetching_nullifiers_range.block_range().start
1038 || block_bound >= full_refetching_nullifiers_range.block_range().end
1039 {
1040 continue;
1041 }
1042 if wallet.get_wallet_block(block_bound).is_err() {
1043 missing_block_bounds.insert(
1044 block_bound,
1045 WalletBlock::from_compact_block(
1046 consensus_parameters,
1047 fetch_request_sender.clone(),
1048 &client::get_compact_block(
1049 fetch_request_sender.clone(),
1050 block_bound,
1051 )
1052 .await?,
1053 )
1054 .await?,
1055 );
1056 }
1057 }
1058 if !missing_block_bounds.is_empty() {
1059 wallet
1060 .append_wallet_blocks(missing_block_bounds)
1061 .map_err(SyncError::WalletError)?;
1062 }
1063 }
1064
1065 let first_unscanned_range = wallet
1066 .get_sync_state()
1067 .map_err(SyncError::WalletError)?
1068 .scan_ranges
1069 .iter()
1070 .find(|scan_range| scan_range.priority() != ScanPriority::Scanned)
1071 .expect("the scan range being processed is not yet set to scanned so at least one unscanned range must exist");
1072 if !first_unscanned_range
1073 .block_range()
1074 .contains(&scan_range.block_range().start)
1075 || !first_unscanned_range
1076 .block_range()
1077 .contains(&(scan_range.block_range().end - 1))
1078 {
1079 state::reset_refetching_nullifiers_scan_range(
1083 wallet
1084 .get_sync_state_mut()
1085 .map_err(SyncError::WalletError)?,
1086 scan_range.block_range().clone(),
1087 );
1088 tracing::debug!(
1089 "Nullifiers discarded and will be re-fetched to avoid missing spends."
1090 );
1091
1092 return Ok(());
1093 }
1094
1095 spend::update_shielded_spends(
1096 consensus_parameters,
1097 wallet,
1098 fetch_request_sender.clone(),
1099 ufvks,
1100 &scanned_blocks,
1101 Some(&mut nullifiers),
1102 )
1103 .await?;
1104
1105 state::set_scanned_scan_range(
1106 wallet
1107 .get_sync_state_mut()
1108 .map_err(SyncError::WalletError)?,
1109 scan_range.block_range().clone(),
1110 true, );
1112 } else {
1113 if !*nullifier_map_limit_exceeded {
1115 let nullifier_map = wallet.get_nullifiers().map_err(SyncError::WalletError)?;
1116 if max_nullifier_map_size(performance_level).is_some_and(|max| {
1117 nullifier_map.orchard.len()
1118 + nullifier_map.sapling.len()
1119 + nullifiers.orchard.len()
1120 + nullifiers.sapling.len()
1121 > max
1122 }) {
1123 *nullifier_map_limit_exceeded = true;
1124 }
1125 }
1126 let mut map_nullifiers = !*nullifier_map_limit_exceeded;
1127
1128 let map_outpoints = scan_range.priority() >= ScanPriority::FoundNote;
1131
1132 for query_scan_range in wallet
1139 .get_sync_state()
1140 .map_err(SyncError::WalletError)?
1141 .scan_ranges()
1142 {
1143 let scan_priority = query_scan_range.priority();
1144 if scan_priority != ScanPriority::Scanned
1145 && scan_priority != ScanPriority::Scanning
1146 && scan_priority != ScanPriority::RefetchingNullifiers
1147 {
1148 break;
1149 }
1150
1151 if scan_priority == ScanPriority::Scanning
1152 && query_scan_range
1153 .block_range()
1154 .contains(&scan_range.block_range().start)
1155 && query_scan_range
1156 .block_range()
1157 .contains(&(scan_range.block_range().end - 1))
1158 {
1159 map_nullifiers = true;
1160 break;
1161 }
1162 }
1163
1164 update_wallet_data(
1165 consensus_parameters,
1166 wallet,
1167 fetch_request_sender.clone(),
1168 ufvks,
1169 &scan_range,
1170 if map_nullifiers {
1171 Some(&mut nullifiers)
1172 } else {
1173 None
1174 },
1175 if map_outpoints {
1176 Some(&mut outpoints)
1177 } else {
1178 None
1179 },
1180 wallet_transactions,
1181 sapling_located_trees,
1182 orchard_located_trees,
1183 )
1184 .await?;
1185 spend::update_transparent_spends(
1186 wallet,
1187 if map_outpoints {
1188 None
1189 } else {
1190 Some(&mut outpoints)
1191 },
1192 )
1193 .map_err(SyncError::WalletError)?;
1194 spend::update_shielded_spends(
1195 consensus_parameters,
1196 wallet,
1197 fetch_request_sender,
1198 ufvks,
1199 &scanned_blocks,
1200 if map_nullifiers {
1201 None
1202 } else {
1203 Some(&mut nullifiers)
1204 },
1205 )
1206 .await?;
1207 add_scanned_blocks(wallet, scanned_blocks, &scan_range)
1208 .map_err(SyncError::WalletError)?;
1209
1210 state::set_scanned_scan_range(
1211 wallet
1212 .get_sync_state_mut()
1213 .map_err(SyncError::WalletError)?,
1214 scan_range.block_range().clone(),
1215 map_nullifiers,
1216 );
1217 state::merge_scan_ranges(
1218 wallet
1219 .get_sync_state_mut()
1220 .map_err(SyncError::WalletError)?,
1221 ScanPriority::ScannedWithoutMapping,
1222 );
1223 }
1224
1225 state::merge_scan_ranges(
1226 wallet
1227 .get_sync_state_mut()
1228 .map_err(SyncError::WalletError)?,
1229 ScanPriority::Scanned,
1230 );
1231 remove_irrelevant_data(wallet).map_err(SyncError::WalletError)?;
1232 tracing::debug!("Scan results processed.");
1233 }
1234 Err(ScanError::ContinuityError(ContinuityError::HashDiscontinuity { height, .. })) => {
1235 tracing::warn!("Hash discontinuity detected before block {height}.");
1236 if height == scan_range.block_range().start
1237 && scan_range.priority() == ScanPriority::Verify
1238 {
1239 tracing::info!("Re-org detected.");
1240 let sync_state = wallet
1241 .get_sync_state_mut()
1242 .map_err(SyncError::WalletError)?;
1243 let last_known_chain_height = sync_state
1244 .last_known_chain_height()
1245 .expect("scan ranges should be non-empty in this scope");
1246
1247 state::set_scan_priority(
1249 sync_state,
1250 scan_range.block_range(),
1251 ScanPriority::Verify,
1252 );
1253
1254 let current_reorg_detection_start_height = state::set_verify_scan_range(
1256 sync_state,
1257 height - 1,
1258 state::VerifyEnd::VerifyHighest,
1259 )
1260 .block_range()
1261 .start;
1262 state::merge_scan_ranges(sync_state, ScanPriority::Verify);
1263
1264 if initial_reorg_detection_start_height - current_reorg_detection_start_height
1265 > MAX_REORG_ALLOWANCE
1266 {
1267 clear_wallet_data(wallet)?;
1268
1269 return Err(ServerError::ChainVerificationError.into());
1270 }
1271
1272 truncate_wallet_data(wallet, current_reorg_detection_start_height - 1)?;
1273
1274 state::set_initial_state(
1275 consensus_parameters,
1276 fetch_request_sender.clone(),
1277 wallet,
1278 last_known_chain_height,
1279 )
1280 .await?;
1281 } else {
1282 scan_results?;
1283 }
1284 }
1285 Err(e) => return Err(e.into()),
1286 }
1287
1288 Ok(())
1289}
1290
1291async fn process_mempool_transaction<W>(
1295 consensus_parameters: &impl consensus::Parameters,
1296 ufvks: &HashMap<AccountId, UnifiedFullViewingKey>,
1297 wallet: &mut W,
1298 raw_transaction: RawTransaction,
1299) -> Result<(), SyncError<W::Error>>
1300where
1301 W: SyncWallet + SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees,
1302{
1303 let mempool_height = wallet
1305 .get_sync_state()
1306 .map_err(SyncError::WalletError)?
1307 .last_known_chain_height()
1308 .expect("wallet height must exist after sync is initialised")
1309 + 1;
1310
1311 let transaction = zcash_primitives::transaction::Transaction::read(
1312 &raw_transaction.data[..],
1313 consensus::BranchId::for_height(consensus_parameters, mempool_height),
1314 )
1315 .map_err(ServerError::InvalidTransaction)?;
1316
1317 tracing::debug!(
1318 "mempool received txid {} at height {}",
1319 transaction.txid(),
1320 mempool_height
1321 );
1322
1323 if let Some(tx) = wallet
1324 .get_wallet_transactions_mut()
1325 .map_err(SyncError::WalletError)?
1326 .get_mut(&transaction.txid())
1327 {
1328 tx.update_status(
1329 ConfirmationStatus::Mempool(mempool_height),
1330 SystemTime::now()
1331 .duration_since(SystemTime::UNIX_EPOCH)
1332 .expect("infalliable for such long time periods")
1333 .as_secs() as u32,
1334 );
1335
1336 return Ok(());
1337 }
1338
1339 scan_pending_transaction(
1340 consensus_parameters,
1341 ufvks,
1342 wallet,
1343 transaction,
1344 ConfirmationStatus::Mempool(mempool_height),
1345 SystemTime::now()
1346 .duration_since(SystemTime::UNIX_EPOCH)
1347 .expect("infalliable for such long time periods")
1348 .as_secs() as u32,
1349 )?;
1350
1351 Ok(())
1352}
1353
1354fn truncate_wallet_data<W>(
1356 wallet: &mut W,
1357 truncate_height: BlockHeight,
1358) -> Result<(), SyncError<W::Error>>
1359where
1360 W: SyncWallet + SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees,
1361{
1362 let sync_state = wallet
1363 .get_sync_state_mut()
1364 .map_err(SyncError::WalletError)?;
1365 let highest_scanned_height = sync_state
1366 .highest_scanned_height()
1367 .expect("should be non-empty in this scope");
1368 let wallet_birthday = sync_state
1369 .wallet_birthday()
1370 .expect("should be non-empty in this scope");
1371 let checked_truncate_height = match truncate_height.cmp(&wallet_birthday) {
1372 std::cmp::Ordering::Greater | std::cmp::Ordering::Equal => truncate_height,
1373 std::cmp::Ordering::Less => consensus::H0,
1374 };
1375
1376 if checked_truncate_height > highest_scanned_height {
1377 return Ok(());
1378 }
1379
1380 wallet
1381 .truncate_wallet_blocks(checked_truncate_height)
1382 .map_err(SyncError::WalletError)?;
1383 wallet
1384 .truncate_wallet_transactions(checked_truncate_height)
1385 .map_err(SyncError::WalletError)?;
1386 wallet
1387 .truncate_nullifiers(checked_truncate_height)
1388 .map_err(SyncError::WalletError)?;
1389 wallet
1390 .truncate_outpoints(checked_truncate_height)
1391 .map_err(SyncError::WalletError)?;
1392 match wallet.truncate_shard_trees(checked_truncate_height) {
1393 Ok(_) => Ok(()),
1394 Err(SyncError::TruncationError(height, pooltype)) => {
1395 clear_wallet_data(wallet)?;
1396
1397 Err(SyncError::TruncationError(height, pooltype))
1398 }
1399 Err(e) => Err(e),
1400 }?;
1401
1402 Ok(())
1403}
1404
1405fn clear_wallet_data<W>(wallet: &mut W) -> Result<(), SyncError<W::Error>>
1406where
1407 W: SyncWallet + SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees,
1408{
1409 let scan_targets = wallet
1410 .get_wallet_transactions()
1411 .map_err(SyncError::WalletError)?
1412 .values()
1413 .filter_map(|transaction| {
1414 transaction
1415 .status()
1416 .get_confirmed_height()
1417 .map(|height| ScanTarget {
1418 block_height: height,
1419 txid: transaction.txid(),
1420 narrow_scan_area: true,
1421 })
1422 })
1423 .collect::<Vec<_>>();
1424 truncate_wallet_data(wallet, consensus::H0)?;
1425 truncate_scan_ranges(
1426 consensus::H0,
1427 wallet
1428 .get_sync_state_mut()
1429 .map_err(SyncError::WalletError)?,
1430 );
1431 wallet
1432 .get_wallet_transactions_mut()
1433 .map_err(SyncError::WalletError)?
1434 .clear();
1435 let sync_state = wallet
1436 .get_sync_state_mut()
1437 .map_err(SyncError::WalletError)?;
1438 add_scan_targets(sync_state, &scan_targets);
1439 wallet.set_save_flag().map_err(SyncError::WalletError)?;
1440
1441 Ok(())
1442}
1443
1444#[allow(clippy::too_many_arguments)]
1446async fn update_wallet_data<W>(
1447 consensus_parameters: &impl consensus::Parameters,
1448 wallet: &mut W,
1449 fetch_request_sender: mpsc::UnboundedSender<FetchRequest>,
1450 ufvks: &HashMap<AccountId, UnifiedFullViewingKey>,
1451 scan_range: &ScanRange,
1452 nullifiers: Option<&mut NullifierMap>,
1453 outpoints: Option<&mut BTreeMap<OutputId, ScanTarget>>,
1454 mut transactions: HashMap<TxId, WalletTransaction>,
1455 sapling_located_trees: Vec<LocatedTreeData<sapling_crypto::Node>>,
1456 orchard_located_trees: Vec<LocatedTreeData<MerkleHashOrchard>>,
1457) -> Result<(), SyncError<W::Error>>
1458where
1459 W: SyncBlocks + SyncTransactions + SyncNullifiers + SyncOutPoints + SyncShardTrees + Send,
1460{
1461 let sync_state = wallet
1462 .get_sync_state_mut()
1463 .map_err(SyncError::WalletError)?;
1464 let highest_scanned_height = sync_state
1465 .highest_scanned_height()
1466 .expect("scan ranges should not be empty in this scope");
1467 for transaction in transactions.values() {
1468 state::update_found_note_shard_priority(
1469 consensus_parameters,
1470 sync_state,
1471 ShieldedProtocol::Sapling,
1472 transaction,
1473 );
1474 state::update_found_note_shard_priority(
1475 consensus_parameters,
1476 sync_state,
1477 ShieldedProtocol::Orchard,
1478 transaction,
1479 );
1480 }
1481 let refetch_nullifier_ranges = {
1489 let block_ranges: Vec<Range<BlockHeight>> = sync_state
1490 .scan_ranges()
1491 .iter()
1492 .filter(|&scan_range| {
1493 scan_range.priority() == ScanPriority::ScannedWithoutMapping
1494 || scan_range.priority() == ScanPriority::RefetchingNullifiers
1495 })
1496 .map(|scan_range| scan_range.block_range().clone())
1497 .collect();
1498
1499 block_ranges
1500 [block_ranges.partition_point(|range| range.start < scan_range.block_range().end)..]
1501 .to_vec()
1502 };
1503 for transaction in transactions.values_mut() {
1504 for note in transaction.sapling_notes.as_mut_slice() {
1505 note.refetch_nullifier_ranges = refetch_nullifier_ranges.clone();
1506 }
1507 for note in transaction.orchard_notes.as_mut_slice() {
1508 note.refetch_nullifier_ranges = refetch_nullifier_ranges.clone();
1509 }
1510 }
1511 for transaction in transactions.values() {
1512 discover_unified_addresses(wallet, ufvks, transaction).map_err(SyncError::WalletError)?;
1513 }
1514
1515 wallet
1516 .extend_wallet_transactions(transactions)
1517 .map_err(SyncError::WalletError)?;
1518 if let Some(nullifiers) = nullifiers {
1519 wallet
1520 .append_nullifiers(nullifiers)
1521 .map_err(SyncError::WalletError)?;
1522 }
1523 if let Some(outpoints) = outpoints {
1524 wallet
1525 .append_outpoints(outpoints)
1526 .map_err(SyncError::WalletError)?;
1527 }
1528 wallet
1529 .update_shard_trees(
1530 fetch_request_sender,
1531 scan_range,
1532 highest_scanned_height,
1533 sapling_located_trees,
1534 orchard_located_trees,
1535 )
1536 .await?;
1537
1538 Ok(())
1539}
1540
1541fn discover_unified_addresses<W>(
1542 wallet: &mut W,
1543 ufvks: &HashMap<AccountId, UnifiedFullViewingKey>,
1544 transaction: &WalletTransaction,
1545) -> Result<(), W::Error>
1546where
1547 W: SyncWallet,
1548{
1549 for note in transaction
1550 .orchard_notes()
1551 .iter()
1552 .filter(|¬e| note.key_id().scope == zip32::Scope::External)
1553 {
1554 let ivk = ufvks
1555 .get(¬e.key_id().account_id())
1556 .expect("ufvk must exist to decrypt this note")
1557 .orchard()
1558 .expect("fvk must exist to decrypt this note")
1559 .to_ivk(zip32::Scope::External);
1560
1561 wallet.add_orchard_address(
1562 note.key_id().account_id(),
1563 note.note().recipient(),
1564 ivk.diversifier_index(¬e.note().recipient())
1565 .expect("must be key used to create this address"),
1566 )?;
1567 }
1568 for note in transaction
1569 .sapling_notes()
1570 .iter()
1571 .filter(|¬e| note.key_id().scope == zip32::Scope::External)
1572 {
1573 let ivk = ufvks
1574 .get(¬e.key_id().account_id())
1575 .expect("ufvk must exist to decrypt this note")
1576 .sapling()
1577 .expect("fvk must exist to decrypt this note")
1578 .to_external_ivk();
1579
1580 wallet.add_sapling_address(
1581 note.key_id().account_id(),
1582 note.note().recipient(),
1583 ivk.decrypt_diversifier(¬e.note().recipient())
1584 .expect("must be key used to create this address"),
1585 )?;
1586 }
1587
1588 Ok(())
1589}
1590
1591fn remove_irrelevant_data<W>(wallet: &mut W) -> Result<(), W::Error>
1592where
1593 W: SyncWallet + SyncBlocks + SyncOutPoints + SyncNullifiers + SyncTransactions,
1594{
1595 let fully_scanned_height = wallet
1596 .get_sync_state()?
1597 .fully_scanned_height()
1598 .expect("scan ranges must be non-empty");
1599
1600 wallet
1601 .get_outpoints_mut()?
1602 .retain(|_, scan_target| scan_target.block_height > fully_scanned_height);
1603 wallet
1604 .get_nullifiers_mut()?
1605 .sapling
1606 .retain(|_, scan_target| scan_target.block_height > fully_scanned_height);
1607 wallet
1608 .get_nullifiers_mut()?
1609 .orchard
1610 .retain(|_, scan_target| scan_target.block_height > fully_scanned_height);
1611 wallet
1612 .get_sync_state_mut()?
1613 .scan_targets
1614 .retain(|scan_target| scan_target.block_height > fully_scanned_height);
1615 remove_irrelevant_blocks(wallet)?;
1616
1617 Ok(())
1618}
1619
1620fn remove_irrelevant_blocks<W>(wallet: &mut W) -> Result<(), W::Error>
1621where
1622 W: SyncWallet + SyncBlocks + SyncTransactions,
1623{
1624 let sync_state = wallet.get_sync_state()?;
1625 let highest_scanned_height = sync_state
1626 .highest_scanned_height()
1627 .expect("should be non-empty");
1628 let scanned_range_bounds = sync_state
1629 .scan_ranges()
1630 .iter()
1631 .filter(|scan_range| {
1632 scan_range.priority() == ScanPriority::Scanned
1633 || scan_range.priority() == ScanPriority::ScannedWithoutMapping
1634 || scan_range.priority() == ScanPriority::RefetchingNullifiers
1635 })
1636 .flat_map(|scanned_range| {
1637 vec![
1638 scanned_range.block_range().start,
1639 scanned_range.block_range().end - 1,
1640 ]
1641 })
1642 .collect::<Vec<_>>();
1643 let wallet_transaction_heights = wallet
1644 .get_wallet_transactions()?
1645 .values()
1646 .filter_map(|tx| tx.status().get_confirmed_height())
1647 .collect::<Vec<_>>();
1648
1649 wallet.get_wallet_blocks_mut()?.retain(|height, _| {
1650 *height >= highest_scanned_height.saturating_sub(MAX_REORG_ALLOWANCE)
1651 || scanned_range_bounds.contains(height)
1652 || wallet_transaction_heights.contains(height)
1653 });
1654
1655 Ok(())
1656}
1657
1658fn add_scanned_blocks<W>(
1659 wallet: &mut W,
1660 mut scanned_blocks: BTreeMap<BlockHeight, WalletBlock>,
1661 scan_range: &ScanRange,
1662) -> Result<(), W::Error>
1663where
1664 W: SyncWallet + SyncBlocks + SyncTransactions,
1665{
1666 let sync_state = wallet.get_sync_state()?;
1667 let highest_scanned_height = sync_state
1668 .highest_scanned_height()
1669 .expect("scan ranges must be non-empty");
1670
1671 let wallet_transaction_heights = wallet
1672 .get_wallet_transactions()?
1673 .values()
1674 .filter_map(|tx| tx.status().get_confirmed_height())
1675 .collect::<Vec<_>>();
1676
1677 scanned_blocks.retain(|height, _| {
1678 *height >= highest_scanned_height.saturating_sub(MAX_REORG_ALLOWANCE)
1679 || *height == scan_range.block_range().start
1680 || *height == scan_range.block_range().end - 1
1681 || wallet_transaction_heights.contains(height)
1682 });
1683
1684 wallet.append_wallet_blocks(scanned_blocks)?;
1685
1686 Ok(())
1687}
1688
1689#[cfg(not(feature = "darkside_test"))]
1690async fn update_subtree_roots<W>(
1691 consensus_parameters: &impl consensus::Parameters,
1692 fetch_request_sender: mpsc::UnboundedSender<FetchRequest>,
1693 wallet: &mut W,
1694) -> Result<(), SyncError<W::Error>>
1695where
1696 W: SyncWallet + SyncShardTrees,
1697{
1698 let sapling_start_index = wallet
1699 .get_shard_trees()
1700 .map_err(SyncError::WalletError)?
1701 .sapling
1702 .store()
1703 .get_shard_roots()
1704 .expect("infallible")
1705 .len() as u32;
1706 let orchard_start_index = wallet
1707 .get_shard_trees()
1708 .map_err(SyncError::WalletError)?
1709 .orchard
1710 .store()
1711 .get_shard_roots()
1712 .expect("infallible")
1713 .len() as u32;
1714 let (sapling_subtree_roots, orchard_subtree_roots) = futures::join!(
1715 client::get_subtree_roots(fetch_request_sender.clone(), sapling_start_index, 0, 0),
1716 client::get_subtree_roots(fetch_request_sender, orchard_start_index, 1, 0)
1717 );
1718
1719 let sapling_subtree_roots = sapling_subtree_roots?;
1720 let orchard_subtree_roots = orchard_subtree_roots?;
1721
1722 let sync_state = wallet
1723 .get_sync_state_mut()
1724 .map_err(SyncError::WalletError)?;
1725 state::add_shard_ranges(
1726 consensus_parameters,
1727 ShieldedProtocol::Sapling,
1728 sync_state,
1729 &sapling_subtree_roots,
1730 );
1731 state::add_shard_ranges(
1732 consensus_parameters,
1733 ShieldedProtocol::Orchard,
1734 sync_state,
1735 &orchard_subtree_roots,
1736 );
1737
1738 let shard_trees = wallet
1739 .get_shard_trees_mut()
1740 .map_err(SyncError::WalletError)?;
1741 witness::add_subtree_roots(
1742 sapling_start_index as usize,
1743 sapling_subtree_roots,
1744 &mut shard_trees.sapling,
1745 )?;
1746 witness::add_subtree_roots(
1747 orchard_start_index as usize,
1748 orchard_subtree_roots,
1749 &mut shard_trees.orchard,
1750 )?;
1751
1752 Ok(())
1753}
1754
1755async fn add_initial_frontier<W>(
1756 consensus_parameters: &impl consensus::Parameters,
1757 fetch_request_sender: mpsc::UnboundedSender<FetchRequest>,
1758 wallet: &mut W,
1759) -> Result<(), SyncError<W::Error>>
1760where
1761 W: SyncWallet + SyncShardTrees,
1762{
1763 let birthday = wallet.get_birthday().map_err(SyncError::WalletError)?;
1764 if birthday
1765 == consensus_parameters
1766 .activation_height(consensus::NetworkUpgrade::Sapling)
1767 .expect("sapling activation height should always return Some")
1768 {
1769 return Ok(());
1770 }
1771
1772 let shard_trees = wallet
1775 .get_shard_trees_mut()
1776 .map_err(SyncError::WalletError)?;
1777 if shard_trees
1778 .sapling
1779 .store()
1780 .checkpoint_count()
1781 .expect("infallible")
1782 == 1
1783 {
1784 let frontiers = client::get_frontiers(fetch_request_sender, birthday).await?;
1785 shard_trees
1786 .sapling
1787 .insert_frontier(
1788 frontiers.final_sapling_tree().clone(),
1789 Retention::Checkpoint {
1790 id: birthday,
1791 marking: Marking::None,
1792 },
1793 )
1794 .expect("infallible");
1795 shard_trees
1796 .orchard
1797 .insert_frontier(
1798 frontiers.final_orchard_tree().clone(),
1799 Retention::Checkpoint {
1800 id: birthday,
1801 marking: Marking::None,
1802 },
1803 )
1804 .expect("infallible");
1805 }
1806
1807 Ok(())
1808}
1809
1810async fn mempool_monitor<C>(
1815 mut client: C,
1816 mempool_transaction_sender: mpsc::Sender<RawTransaction>,
1817 unprocessed_transactions_count: Arc<AtomicU8>,
1818 shutdown_mempool: Arc<AtomicBool>,
1819) -> Result<(), MempoolError>
1820where
1821 C: Clone + Indexer + TransparentIndexer + Sync + Send + 'static,
1822{
1823 let mut interval = tokio::time::interval(Duration::from_secs(1));
1824 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
1825 'main: loop {
1826 let response =
1827 client::get_mempool_transaction_stream(&mut client, shutdown_mempool.clone()).await;
1828
1829 match response {
1830 Ok(mut mempool_stream) => {
1831 interval.reset();
1832 loop {
1833 tokio::select! {
1834 mempool_stream_message = mempool_stream.message() => {
1835 match mempool_stream_message.unwrap_or(None) {
1836 Some(raw_transaction) => {
1837 let _ignore_error = mempool_transaction_sender
1838 .send(raw_transaction)
1839 .await;
1840 unprocessed_transactions_count.fetch_add(1, atomic::Ordering::Release);
1841 }
1842 None => {
1843 continue 'main;
1844 }
1845 }
1846
1847 }
1848
1849 _ = interval.tick() => {
1850 if shutdown_mempool.load(atomic::Ordering::Acquire) {
1851 break 'main;
1852 }
1853 }
1854 }
1855 }
1856 }
1857 Err(e @ MempoolError::ShutdownWithoutStream) => return Err(e),
1858 Err(MempoolError::ServerError(e)) => {
1859 tracing::warn!("Mempool stream request failed! Status: {e}.\nRetrying...");
1860 tokio::time::sleep(Duration::from_secs(3)).await;
1861 }
1862 }
1863 }
1864
1865 Ok(())
1866}
1867
1868fn expire_transactions<W>(wallet: &mut W) -> Result<(), SyncError<W::Error>>
1870where
1871 W: SyncWallet + SyncTransactions,
1872{
1873 let last_known_chain_height = wallet
1874 .get_sync_state()
1875 .map_err(SyncError::WalletError)?
1876 .last_known_chain_height()
1877 .expect("wallet height must exist after scan ranges have been updated");
1878 let wallet_transactions = wallet
1879 .get_wallet_transactions_mut()
1880 .map_err(SyncError::WalletError)?;
1881
1882 let expired_txids = wallet_transactions
1883 .values()
1884 .filter(|transaction| {
1885 transaction.status().is_pending()
1886 && last_known_chain_height >= transaction.transaction().expiry_height()
1887 })
1888 .map(super::wallet::WalletTransaction::txid)
1889 .collect::<Vec<_>>();
1890 set_transactions_failed(wallet_transactions, expired_txids);
1891
1892 Ok(())
1893}
1894
1895fn max_nullifier_map_size(performance_level: PerformanceLevel) -> Option<usize> {
1896 match performance_level {
1897 PerformanceLevel::Low => Some(0),
1898 PerformanceLevel::Medium => Some(125_000),
1899 PerformanceLevel::High => Some(2_000_000),
1900 PerformanceLevel::Maximum => None,
1901 }
1902}
1903
1904#[cfg(test)]
1905mod test {
1906 mod checked_height_validation {
1907 use zcash_protocol::consensus::BlockHeight;
1908 use zcash_protocol::local_consensus::LocalNetwork;
1909 const LOCAL_NETWORK: LocalNetwork = LocalNetwork {
1910 overwinter: Some(BlockHeight::from_u32(1)),
1911 sapling: Some(BlockHeight::from_u32(3)),
1912 blossom: Some(BlockHeight::from_u32(3)),
1913 heartwood: Some(BlockHeight::from_u32(3)),
1914 canopy: Some(BlockHeight::from_u32(3)),
1915 nu5: Some(BlockHeight::from_u32(3)),
1916 nu6: Some(BlockHeight::from_u32(3)),
1917 nu6_1: Some(BlockHeight::from_u32(3)),
1918 nu6_2: Some(BlockHeight::from_u32(3)),
1919 };
1920 use crate::{error::SyncError, mocks::MockWalletError, sync::checked_wallet_height};
1921 #[tokio::test]
1924 async fn get_sync_state_error() {
1925 let builder = crate::mocks::MockWalletBuilder::new();
1926 let test_error = "get_sync_state_error";
1927 let mut test_wallet = builder
1928 .get_sync_state_patch(Box::new(|_| {
1929 Err(MockWalletError::AnErrorVariant(test_error.to_string()))
1930 }))
1931 .create_mock_wallet();
1932 let res =
1933 checked_wallet_height(&mut test_wallet, BlockHeight::from_u32(1), &LOCAL_NETWORK);
1934 assert!(matches!(
1935 res,
1936 Err(SyncError::WalletError(
1937 crate::mocks::MockWalletError::AnErrorVariant(ref s)
1938 )) if s == test_error
1939 ));
1940 }
1941
1942 mod last_known_chain_height {
1943 use crate::{
1944 sync::{MAX_REORG_ALLOWANCE, ScanRange},
1945 wallet::SyncState,
1946 };
1947 const DEFAULT_START_HEIGHT: BlockHeight = BlockHeight::from_u32(1);
1948 const _DEFAULT_LAST_KNOWN_HEIGHT: BlockHeight = BlockHeight::from_u32(102);
1949 const DEFAULT_CHAIN_HEIGHT: BlockHeight = BlockHeight::from_u32(110);
1950
1951 use super::*;
1952 #[tokio::test]
1953 async fn above_allowance() {
1954 const LAST_KNOWN_HEIGHT: BlockHeight = BlockHeight::from_u32(211);
1955 let lkch = vec![ScanRange::from_parts(
1956 DEFAULT_START_HEIGHT..LAST_KNOWN_HEIGHT,
1957 crate::sync::ScanPriority::Scanned,
1958 )];
1959 let state = SyncState {
1960 scan_ranges: lkch,
1961 ..Default::default()
1962 };
1963 let builder = crate::mocks::MockWalletBuilder::new();
1964 let mut test_wallet = builder.sync_state(state).create_mock_wallet();
1965 let res =
1966 checked_wallet_height(&mut test_wallet, DEFAULT_CHAIN_HEIGHT, &LOCAL_NETWORK);
1967 if let Err(e) = res {
1968 assert_eq!(
1969 e.to_string(),
1970 format!(
1971 "wallet height {} is more than {} blocks ahead of best chain height {}",
1972 LAST_KNOWN_HEIGHT - 1,
1973 MAX_REORG_ALLOWANCE,
1974 DEFAULT_CHAIN_HEIGHT
1975 )
1976 );
1977 } else {
1978 panic!()
1979 }
1980 }
1981 #[tokio::test]
1982 async fn above_chain_height_below_allowance() {
1983 let lkch = vec![ScanRange::from_parts(
1987 BlockHeight::from_u32(6)..BlockHeight::from_u32(10),
1988 crate::sync::ScanPriority::Scanned,
1989 )];
1990 let state = SyncState {
1991 scan_ranges: lkch,
1992 ..Default::default()
1993 };
1994 let builder = crate::mocks::MockWalletBuilder::new();
1995 let mut test_wallet = builder.sync_state(state).create_mock_wallet();
1996 let chain_height = BlockHeight::from_u32(4);
1997 let res = checked_wallet_height(&mut test_wallet, chain_height, &LOCAL_NETWORK);
2002 assert_eq!(res.unwrap(), BlockHeight::from_u32(4));
2003 }
2004 #[ignore = "in progress"]
2005 #[tokio::test]
2006 async fn equal_or_below_chain_height_and_above_sapling() {
2007 let lkch = vec![ScanRange::from_parts(
2008 BlockHeight::from_u32(1)..BlockHeight::from_u32(10),
2009 crate::sync::ScanPriority::Scanned,
2010 )];
2011 let state = SyncState {
2012 scan_ranges: lkch,
2013 ..Default::default()
2014 };
2015 let builder = crate::mocks::MockWalletBuilder::new();
2016 let mut _test_wallet = builder.sync_state(state).create_mock_wallet();
2017 }
2018 #[ignore = "in progress"]
2019 #[tokio::test]
2020 async fn equal_or_below_chain_height_and_below_sapling() {
2021 let lkch = vec![ScanRange::from_parts(
2024 BlockHeight::from_u32(1)..BlockHeight::from_u32(10),
2025 crate::sync::ScanPriority::Scanned,
2026 )];
2027 let state = SyncState {
2028 scan_ranges: lkch,
2029 ..Default::default()
2030 };
2031 let builder = crate::mocks::MockWalletBuilder::new();
2032 let mut _test_wallet = builder.sync_state(state).create_mock_wallet();
2033 }
2034 #[ignore = "in progress"]
2035 #[tokio::test]
2036 async fn below_sapling() {
2037 let lkch = vec![ScanRange::from_parts(
2038 BlockHeight::from_u32(1)..BlockHeight::from_u32(10),
2039 crate::sync::ScanPriority::Scanned,
2040 )];
2041 let state = SyncState {
2042 scan_ranges: lkch,
2043 ..Default::default()
2044 };
2045 let builder = crate::mocks::MockWalletBuilder::new();
2046 let mut _test_wallet = builder.sync_state(state).create_mock_wallet();
2047 }
2048 }
2049 mod no_last_known_chain_height {
2050 use super::*;
2051 #[tokio::test]
2053 async fn get_bday_error() {
2054 let test_error = "get_bday_error";
2055 let builder = crate::mocks::MockWalletBuilder::new();
2056 let mut test_wallet = builder
2057 .get_birthday_patch(Box::new(|_| {
2058 Err(crate::mocks::MockWalletError::AnErrorVariant(
2059 test_error.to_string(),
2060 ))
2061 }))
2062 .create_mock_wallet();
2063 let res = checked_wallet_height(
2064 &mut test_wallet,
2065 BlockHeight::from_u32(1),
2066 &LOCAL_NETWORK,
2067 );
2068 assert!(matches!(
2069 res,
2070 Err(SyncError::WalletError(
2071 crate::mocks::MockWalletError::AnErrorVariant(ref s)
2072 )) if s == test_error
2073 ));
2074 }
2075 #[ignore = "in progress"]
2076 #[tokio::test]
2077 async fn raw_bday_above_chain_height() {
2078 let builder = crate::mocks::MockWalletBuilder::new();
2079 let mut test_wallet = builder
2080 .birthday(BlockHeight::from_u32(15))
2081 .create_mock_wallet();
2082 let res = checked_wallet_height(
2083 &mut test_wallet,
2084 BlockHeight::from_u32(1),
2085 &LOCAL_NETWORK,
2086 );
2087 if let Err(e) = res {
2088 assert_eq!(
2089 e.to_string(),
2090 format!(
2091 "wallet height is more than {} blocks ahead of best chain height",
2092 15 - 1
2093 )
2094 );
2095 } else {
2096 panic!()
2097 }
2098 }
2099 mod sapling_height {
2100 use super::*;
2101 #[tokio::test]
2102 async fn raw_bday_above() {
2103 let builder = crate::mocks::MockWalletBuilder::new();
2104 let mut test_wallet = builder
2105 .birthday(BlockHeight::from_u32(4))
2106 .create_mock_wallet();
2107 let res = checked_wallet_height(
2108 &mut test_wallet,
2109 BlockHeight::from_u32(5),
2110 &LOCAL_NETWORK,
2111 );
2112 assert_eq!(res.unwrap(), BlockHeight::from_u32(4 - 1));
2113 }
2114 #[tokio::test]
2115 async fn raw_bday_equal() {
2116 let builder = crate::mocks::MockWalletBuilder::new();
2117 let mut test_wallet = builder
2118 .birthday(BlockHeight::from_u32(3))
2119 .create_mock_wallet();
2120 let res = checked_wallet_height(
2121 &mut test_wallet,
2122 BlockHeight::from_u32(5),
2123 &LOCAL_NETWORK,
2124 );
2125 assert_eq!(res.unwrap(), BlockHeight::from_u32(3 - 1));
2126 }
2127 #[tokio::test]
2128 async fn raw_bday_below() {
2129 let builder = crate::mocks::MockWalletBuilder::new();
2130 let mut test_wallet = builder
2131 .birthday(BlockHeight::from_u32(1))
2132 .create_mock_wallet();
2133 let res = checked_wallet_height(
2134 &mut test_wallet,
2135 BlockHeight::from_u32(5),
2136 &LOCAL_NETWORK,
2137 );
2138 assert!(matches!(res, Err(SyncError::BirthdayBelowSapling(1, 3))));
2139 }
2140 }
2141 }
2142 }
2143}