|
1 | | -//! Filter synchronization and management for the Dash SPV client. |
2 | | -
|
3 | | -use crate::error::{Result, SpvError}; |
4 | | -use crate::network::NetworkManager; |
5 | | -use crate::storage::StorageManager; |
6 | | -use crate::sync::SyncManager; |
7 | | -use crate::types::FilterMatch; |
8 | | -use crate::types::SpvStats; |
9 | | -use key_wallet_manager::wallet_interface::WalletInterface; |
10 | | -use std::sync::Arc; |
11 | | -use tokio::sync::RwLock; |
12 | | - |
13 | | -/// Filter synchronization manager for coordinating filter downloads and checking. |
14 | | -pub struct FilterSyncCoordinator<'a, S: StorageManager, N: NetworkManager, W: WalletInterface> { |
15 | | - sync_manager: &'a mut SyncManager<S, N, W>, |
16 | | - storage: &'a mut S, |
17 | | - network: &'a mut N, |
18 | | - stats: &'a Arc<RwLock<SpvStats>>, |
19 | | - running: &'a Arc<RwLock<bool>>, |
20 | | -} |
21 | | - |
22 | | -impl< |
23 | | - 'a, |
24 | | - S: StorageManager + Send + Sync + 'static, |
25 | | - N: NetworkManager + Send + Sync + 'static, |
26 | | - W: WalletInterface, |
27 | | - > FilterSyncCoordinator<'a, S, N, W> |
28 | | -{ |
29 | | - /// Create a new filter sync coordinator. |
30 | | - pub fn new( |
31 | | - sync_manager: &'a mut SyncManager<S, N, W>, |
32 | | - storage: &'a mut S, |
33 | | - network: &'a mut N, |
34 | | - stats: &'a Arc<RwLock<SpvStats>>, |
35 | | - running: &'a Arc<RwLock<bool>>, |
36 | | - ) -> Self { |
37 | | - Self { |
38 | | - sync_manager, |
39 | | - storage, |
40 | | - network, |
41 | | - stats, |
42 | | - running, |
43 | | - } |
44 | | - } |
45 | | - |
46 | | - /// Sync compact filters for recent blocks and check for matches. |
47 | | - /// Sync and check filters with internal monitoring loop management. |
48 | | - /// This method automatically handles the monitoring loop required for CFilter message processing. |
49 | | - pub async fn sync_and_check_filters_with_monitoring( |
50 | | - &mut self, |
51 | | - num_blocks: Option<u32>, |
52 | | - ) -> Result<Vec<FilterMatch>> { |
53 | | - // Just delegate to the regular method for now - the real fix is in sync_filters_coordinated |
54 | | - self.sync_and_check_filters(num_blocks).await |
55 | | - } |
56 | | - |
57 | | - pub async fn sync_and_check_filters( |
58 | | - &mut self, |
59 | | - num_blocks: Option<u32>, |
60 | | - ) -> Result<Vec<FilterMatch>> { |
61 | | - let running = self.running.read().await; |
62 | | - if !*running { |
63 | | - return Err(SpvError::Config("Client not running".to_string())); |
64 | | - } |
65 | | - drop(running); |
66 | | - |
67 | | - // Get current filter tip height to determine range (use filter headers, not block headers) |
68 | | - // This ensures consistency between range calculation and progress tracking |
69 | | - let tip_height = |
70 | | - self.storage.get_filter_tip_height().await.map_err(SpvError::Storage)?.unwrap_or(0); |
71 | | - |
72 | | - // Determine how many blocks to request |
73 | | - let num_blocks = num_blocks.unwrap_or(100).max(1); |
74 | | - let default_start = tip_height.saturating_sub(num_blocks - 1); |
75 | | - |
76 | | - // Ask the wallet for an earliest rescan height, falling back to the default window. |
77 | | - let wallet_hint = self.sync_manager.wallet_birth_height_hint().await; |
78 | | - let mut start_height = wallet_hint.unwrap_or(default_start).min(default_start); |
79 | | - |
80 | | - // Respect any user-provided start height hint from the configuration. |
81 | | - if let Some(config_start) = self.sync_manager.config_start_height() { |
82 | | - let capped = config_start.min(tip_height); |
83 | | - start_height = start_height.max(capped); |
84 | | - } |
85 | | - |
86 | | - // Make sure we never request past the current tip |
87 | | - start_height = start_height.min(tip_height); |
88 | | - |
89 | | - let actual_count = if start_height <= tip_height { |
90 | | - tip_height - start_height + 1 |
91 | | - } else { |
92 | | - 0 |
93 | | - }; |
94 | | - |
95 | | - tracing::info!( |
96 | | - "Requesting filters from height {} to {} ({} blocks based on filter tip height)", |
97 | | - start_height, |
98 | | - tip_height, |
99 | | - actual_count |
100 | | - ); |
101 | | - if let Some(hint) = wallet_hint { |
102 | | - tracing::debug!("Wallet hint for earliest required height: {}", hint); |
103 | | - } |
104 | | - tracing::info!("Filter processing and matching will happen automatically in background thread as CFilter messages arrive"); |
105 | | - |
106 | | - // Send filter requests - processing will happen automatically in the background |
107 | | - if actual_count > 0 { |
108 | | - self.sync_filters_coordinated(start_height, actual_count).await?; |
109 | | - } else { |
110 | | - tracing::debug!("No filters requested because calculated range is empty"); |
111 | | - } |
112 | | - |
113 | | - // Return empty vector since matching happens asynchronously in the filter processor thread |
114 | | - // Actual matches will be processed and blocks requested automatically when CFilter messages arrive |
115 | | - Ok(Vec::new()) |
116 | | - } |
117 | | - |
118 | | - /// Sync filters for a specific height range. |
119 | | - pub async fn sync_filters_range( |
120 | | - &mut self, |
121 | | - start_height: Option<u32>, |
122 | | - count: Option<u32>, |
123 | | - ) -> Result<()> { |
124 | | - // Get filter tip height to determine default values |
125 | | - let filter_tip_height = |
126 | | - self.storage.get_filter_tip_height().await.map_err(SpvError::Storage)?.unwrap_or(0); |
127 | | - |
128 | | - let start = start_height.unwrap_or(filter_tip_height.saturating_sub(99)); |
129 | | - let num_blocks = count.unwrap_or(100); |
130 | | - |
131 | | - tracing::info!( |
132 | | - "Starting filter sync for specific range from height {} ({} blocks)", |
133 | | - start, |
134 | | - num_blocks |
135 | | - ); |
136 | | - |
137 | | - self.sync_filters_coordinated(start, num_blocks).await |
138 | | - } |
139 | | - |
140 | | - /// Sync filters in coordination with the monitoring loop using flow control processing |
141 | | - async fn sync_filters_coordinated(&mut self, start_height: u32, count: u32) -> Result<()> { |
142 | | - tracing::info!("Starting coordinated filter sync with flow control from height {} to {} ({} filters expected)", |
143 | | - start_height, start_height + count - 1, count); |
144 | | - |
145 | | - // Start tracking filter sync progress |
146 | | - crate::sync::filters::FilterSyncManager::<S, N>::start_filter_sync_tracking( |
147 | | - self.stats, |
148 | | - count as u64, |
149 | | - ) |
150 | | - .await; |
151 | | - |
152 | | - // Use the new flow control method |
153 | | - self.sync_manager |
154 | | - .filter_sync_mut() |
155 | | - .sync_filters_with_flow_control( |
156 | | - &mut *self.network, |
157 | | - &mut *self.storage, |
158 | | - Some(start_height), |
159 | | - Some(count), |
160 | | - ) |
161 | | - .await |
162 | | - .map_err(SpvError::Sync)?; |
163 | | - |
164 | | - let (pending_count, active_count, flow_enabled) = |
165 | | - self.sync_manager.filter_sync().get_flow_control_status(); |
166 | | - tracing::info!("✅ Filter sync with flow control initiated (flow control enabled: {}, {} requests queued, {} active)", |
167 | | - flow_enabled, pending_count, active_count); |
168 | | - |
169 | | - Ok(()) |
170 | | - } |
171 | | -} |
0 commit comments