diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs index 4fe9c7e..e6a27e6 100644 --- a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs @@ -416,6 +416,14 @@ impl LightClientConfig { } } +/// A block range that the wallet has already block-scanned but whose t-address (transparent) scan did not complete, +/// with the block times needed to date the transactions found in it. Memory only; never written to the wallet file. +#[derive(Debug, Clone, Default, PartialEq)] +struct TaddrPending { + start: u64, + times: HashMap, +} + pub struct LightClient { pub wallet : Arc>, @@ -426,6 +434,7 @@ pub struct LightClient { pub sapling_spend : Cow<'static, [u8]>, sync_lock : Mutex<()>, + t_pending : Mutex>, // leaf lock: never held while taking sync_lock or the wallet lock sync_status : Arc>, // The current syncing status of the Wallet. pub shutdown_flag : Arc, // Signal mempool threads to stop } @@ -514,6 +523,7 @@ impl LightClient { sync_lock : Mutex::new(()), sync_status : Arc::new(RwLock::new(WalletStatus::new())), shutdown_flag : Arc::new(AtomicBool::new(false)), + t_pending : Mutex::new(None), }; l.set_wallet_initial_state(0); @@ -543,6 +553,7 @@ impl LightClient { sync_lock : Mutex::new(()), sync_status : Arc::new(RwLock::new(WalletStatus::new())), shutdown_flag : Arc::new(AtomicBool::new(false)), + t_pending : Mutex::new(None), }; l.set_wallet_initial_state(latest_block); @@ -571,6 +582,7 @@ impl LightClient { sync_lock : Mutex::new(()), sync_status : Arc::new(RwLock::new(WalletStatus::new())), shutdown_flag : Arc::new(AtomicBool::new(false)), + t_pending : Mutex::new(None), }; // println!("Setting birthday to {}", birthday); @@ -659,6 +671,7 @@ impl LightClient { sync_lock : Mutex::new(()), sync_status : Arc::new(RwLock::new(WalletStatus::new())), shutdown_flag : Arc::new(AtomicBool::new(false)), + t_pending : Mutex::new(None), }; #[cfg(feature = "embed_params")] @@ -1536,6 +1549,9 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { // otherwise find them gone mid-scan. No caller holds sync_lock, and the order is sync_lock -> wallet. let _sync = self.sync_lock.lock().unwrap(); + // The whole chain is scanned again from `height`: nothing is owed any more. + *self.t_pending.lock().unwrap_or_else(|p| p.into_inner()) = None; + // First, clear the state from the wallet self.wallet.read().unwrap().clear_blocks(); @@ -1571,7 +1587,7 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { let mut retry_count = 0; loop { - match self.do_sync_internal(print_updates, retry_count) { + match self.do_sync_internal(print_updates) { Ok(j) => return Ok(j), Err(e) => { retry_count += 1; @@ -1616,7 +1632,7 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { result.map(|(txid, _)| txid) } - fn do_sync_internal(&self, print_updates: bool, retry_count: u32) -> Result { + fn do_sync_internal(&self, print_updates: bool) -> Result { // We can only do one sync at a time because we sync blocks in serial order // If we allow multiple syncs, they'll all get jumbled up. let _lock = self.sync_lock.lock().unwrap(); @@ -1649,8 +1665,16 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { let scan_batch_size = 25000; let mut end_height = std::cmp::min(last_scanned_height + scan_batch_size, latest_block); - // If there's nothing to scan, just return + // If there's nothing to scan, just return. A t-address scan that failed earlier is still owed even when no + // block is: pay it now, or a transient failure leaves a hole in the transparent history that nothing revisits. if last_scanned_height == latest_block { + if self.taddr_has_pending() { + let pool = ThreadPool::new(max(2, min(8, num_cpus::get()))); + let block_times: Arc>> = Arc::new(RwLock::new(HashMap::new())); + self.taddress_stage(last_scanned_height + 1, last_scanned_height, &block_times, |s, e, bt| { + self.scan_taddress_txids(&pool, bt, s, e, self.config.no_cert_verification).map(|_| ()) + })?; + } info!("Nothing to sync, returning"); return Ok(object!{ "result" => "success" }) } @@ -1680,9 +1704,7 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { .map_err(|e| format!("Error creating runtime: {:?}", e))?; // Fetch CompactBlocks in increments - let mut pass = 0; loop { - pass +=1 ; // Collect all block times, because we'll need to update transparent tx // datetime via the block height timestamp let block_times = Arc::new(RwLock::new(HashMap::new())); @@ -1715,7 +1737,7 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { let last_invalid_height = Arc::new(AtomicI32::new(0)); let last_invalid_height_inner = last_invalid_height.clone(); let tpool = pool.clone(); - fetch_blocks_with_runtime(&mut rt, &self.get_server_uri(), start_height, end_height, self.config.no_cert_verification, pool.clone(), + let fetch_result = fetch_blocks_with_runtime(&mut rt, &self.get_server_uri(), start_height, end_height, self.config.no_cert_verification, pool.clone(), move |encoded_block: &[u8], height: u64| { // Process the block only if there were no previous errors if last_invalid_height_inner.load(Ordering::SeqCst) > 0 { @@ -1745,7 +1767,13 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { }; local_bytes_downloaded.fetch_add(encoded_block.len(), Ordering::SeqCst); - })?; + }); + if let Err(e) = fetch_result { + // Blocks streamed before the failure are already in the wallet and the retry resumes after them, so + // their transparent history has to be scanned later: keep the range and the times seen so far. + self.taddr_note_unscanned(start_height, &block_times); + return Err(e); + } { @@ -1772,6 +1800,9 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { } if invalid_height > 0 { + // Blocks [start_height, invalid_height) of this pass were scanned and are kept, but this pass skips + // its t-scan: owe it (the next pass folds it into its range). + self.taddr_note_unscanned(start_height, &block_times); // Reset the scanning heights last_scanned_height = (invalid_height - 1) as u64; end_height = std::cmp::min(last_scanned_height + scan_batch_size, latest_block); @@ -1785,17 +1816,12 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { // So, reset the total_reorg total_reorg = 0; - // If this is the first pass after a retry, fetch older t address txids too, becuse - // they might have been missed last time. - let transparent_start_height = if pass == 1 && retry_count > 0 { - start_height - scan_batch_size - } else { - start_height - }; - - // Fetch the txids our transparent addresses are involved with. TLS verification follows the - // config (default on); `dangerous` is the only way to turn it off. - self.scan_taddress_txids(&pool, block_times, transparent_start_height, end_height, self.config.no_cert_verification)?; + // Fetch the txids our transparent addresses are involved with, from the start of this pass or of any + // range an earlier failed pass left un-t-scanned (taddress_stage). TLS verification follows the config + // (default on); `dangerous` is the only way to turn it off. + self.taddress_stage(start_height, end_height, &block_times, |s, e, bt| { + self.scan_taddress_txids(&pool, bt, s, e, self.config.no_cert_verification).map(|_| ()) + })?; // Do block height accounting last_scanned_height = end_height; @@ -1834,6 +1860,69 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { } } } + /// Start of the next t-address scan: the lower of this pass's first block and any range an earlier pass left + /// un-t-scanned. The saved block times are folded into `block_times` (entries already there win: after a + /// reorg they are the fresher ones). Never subtracts, so it cannot underflow. + fn taddr_scan_start(&self, pass_start: u64, block_times: &Arc>>) -> u64 { + let pending = self.t_pending.lock().unwrap_or_else(|p| p.into_inner()); + match pending.as_ref() { + None => pass_start, + Some(p) => { + let mut bt = block_times.write().unwrap(); + for (h, t) in p.times.iter() { + bt.entry(*h).or_insert(*t); + } + min(p.start, pass_start) + } + } + } + + /// Record that blocks from `start` on were block-scanned but not t-scanned. Merges with an earlier record + /// (lowest start; this pass's times win). + fn taddr_note_unscanned(&self, start: u64, block_times: &Arc>>) { + let mut times = block_times.read().unwrap().clone(); + let mut pending = self.t_pending.lock().unwrap_or_else(|p| p.into_inner()); + let start = match pending.take() { + Some(old) => { + for (h, t) in old.times { + times.entry(h).or_insert(t); + } + min(old.start, start) + } + None => start, + }; + warn!("t-address scan owed from height {} ({} block times kept); it will be retried", start, times.len()); + *pending = Some(TaddrPending { start, times }); + } + + fn taddr_has_pending(&self) -> bool { + self.t_pending.lock().unwrap_or_else(|p| p.into_inner()).is_some() + } + + /// The t-address stage of one pass: scans [min(pass_start, owed range), end_height] with `scan`. On Err the whole + /// range stays owed (and is kept with its block times), on Ok the debt is cleared. `scan` is a seam for tests. + fn taddress_stage(&self, pass_start: u64, end_height: u64, block_times: &Arc>>, scan: F) + -> Result<(), String> + where F: FnOnce(u64, u64, Arc>>) -> Result<(), String> + { + let t_start = self.taddr_scan_start(pass_start, block_times); + if t_start > end_height { + // Nothing to scan (an owed range above a reorged-down tip): those blocks are fetched again and t-scanned then. + *self.t_pending.lock().unwrap_or_else(|p| p.into_inner()) = None; + return Ok(()); + } + match scan(t_start, end_height, block_times.clone()) { + Ok(()) => { + *self.t_pending.lock().unwrap_or_else(|p| p.into_inner()) = None; + Ok(()) + } + Err(e) => { + self.taddr_note_unscanned(t_start, block_times); + Err(e) + } + } + } + fn scan_taddress_txids( &self, pool: &ThreadPool, @@ -2074,6 +2163,7 @@ pub mod tests { sync_lock : Mutex::new(()), sync_status : Arc::new(RwLock::new(WalletStatus::new())), shutdown_flag : Arc::new(AtomicBool::new(false)), + t_pending : Mutex::new(None), }; { let addresses = lc.do_address(); @@ -2352,6 +2442,139 @@ pub fn test_bad_import() { assert_eq!(one_panic, Err("t-address scan job panicked".to_string())); } + // C242: the extracted t-address scan seam. These drive taddress_stage / taddr_note_unscanned / taddr_scan_start + // directly (the gRPC sync path can't run offline), covering the three original defects: no remembered range, + // discarded block times, and the wrapping rewind subtraction. + type Bt = std::sync::Arc>>; + fn bt_from(pairs: &[(u64, u32)]) -> Bt { + std::sync::Arc::new(std::sync::RwLock::new(pairs.iter().cloned().collect())) + } + + #[test] + pub fn test_taddr_stage_failure_keeps_range_and_times() { + let tmp = TempDir::new("lctest").unwrap(); + let lc = LightClient::new(&main_config(&tmp), 0).unwrap(); + + // Pass 1: block times 101..=200, t-scan fails -> the whole range stays owed with its times. + let bt1: Bt = std::sync::Arc::new(std::sync::RwLock::new( + (101..=200u64).map(|h| (h, h as u32)).collect())); + let r = lc.taddress_stage(101, 200, &bt1, |_s, _e, _bt| Err("boom".to_string())); + assert_eq!(r, Err("boom".to_string())); + assert!(lc.taddr_has_pending()); + + // Pass 2: new block times 201..=300, t-scan succeeds; it must cover [101, 300] with both passes' times. + let bt2: Bt = std::sync::Arc::new(std::sync::RwLock::new( + (201..=300u64).map(|h| (h, h as u32)).collect())); + let seen = std::sync::Arc::new(std::sync::Mutex::new(None)); + let seen_c = seen.clone(); + let r = lc.taddress_stage(201, 300, &bt2, move |s, e, bt| { + let m = bt.read().unwrap(); + *seen_c.lock().unwrap() = Some((s, e, m.contains_key(&150), m.contains_key(&250))); + Ok(()) + }); + assert_eq!(r, Ok(())); + let (s, e, has150, has250) = seen.lock().unwrap().unwrap(); + assert_eq!(s, 101); + assert_eq!(e, 300); + assert!(has150, "the earlier pass's block time must be folded in"); + assert!(has250, "the current pass's block time must be present"); + assert!(!lc.taddr_has_pending(), "success clears the debt"); + } + + #[test] + pub fn test_taddr_stage_no_underflow_below_batch_size() { + let tmp = TempDir::new("lctest").unwrap(); + let lc = LightClient::new(&main_config(&tmp), 0).unwrap(); + + // An owed range below the pass start pulls the scan start down to the owed (lower) height. + lc.taddr_note_unscanned(3, &bt_from(&[(3, 3)])); + let got = std::sync::Arc::new(std::sync::Mutex::new(u64::MAX)); + let got_c = got.clone(); + lc.taddress_stage(10, 1000, &bt_from(&[(10, 10)]), move |s, _e, _bt| { + *got_c.lock().unwrap() = s; + Ok(()) + }).unwrap(); + assert_eq!(*got.lock().unwrap(), 3); + + // No debt: the scan starts exactly at the pass start. For every start in [0, 25000) the old + // `start_height - scan_batch_size` formula wrapped to ~1.8e19; the fix never subtracts. + for ps in 0u64..=24_999 { + let got = std::sync::Arc::new(std::sync::Mutex::new(u64::MAX)); + let got_c = got.clone(); + lc.taddress_stage(ps, ps + 1, &bt_from(&[(ps, 1)]), move |s, _e, _bt| { + *got_c.lock().unwrap() = s; + Ok(()) + }).unwrap(); + let s = *got.lock().unwrap(); + assert!(s <= ps, "scan start {} must never exceed pass start {}", s, ps); + assert_eq!(s, ps); + } + } + + #[test] + pub fn test_taddr_stage_new_times_win_after_reorg() { + let tmp = TempDir::new("lctest").unwrap(); + let lc = LightClient::new(&main_config(&tmp), 0).unwrap(); + + // Owed from a pass that saw height 150 at time 111; the new pass sees 150 at 222 (a reorg re-dated it). + lc.taddr_note_unscanned(150, &bt_from(&[(150, 111)])); + let seen = std::sync::Arc::new(std::sync::Mutex::new(0u32)); + let seen_c = seen.clone(); + lc.taddress_stage(150, 300, &bt_from(&[(150, 222)]), move |_s, _e, bt| { + *seen_c.lock().unwrap() = *bt.read().unwrap().get(&150).unwrap(); + Ok(()) + }).unwrap(); + assert_eq!(*seen.lock().unwrap(), 222, "the current pass's fresher time must win"); + } + + #[test] + pub fn test_taddr_note_unscanned_merges() { + let tmp = TempDir::new("lctest").unwrap(); + let lc = LightClient::new(&main_config(&tmp), 0).unwrap(); + + lc.taddr_note_unscanned(100, &bt_from(&[(100, 1)])); + lc.taddr_note_unscanned(300, &bt_from(&[(300, 3)])); + // taddr_scan_start folds the owed times into its argument without clearing the debt. + let probe = bt_from(&[]); + assert_eq!(lc.taddr_scan_start(10_000, &probe), 100, "the merged start is the lowest seen"); + assert!(probe.read().unwrap().contains_key(&100)); + assert!(probe.read().unwrap().contains_key(&300)); + + lc.taddr_note_unscanned(50, &bt_from(&[(50, 5)])); + let probe2 = bt_from(&[]); + assert_eq!(lc.taddr_scan_start(10_000, &probe2), 50, "a lower note lowers the start"); + assert!(lc.taddr_has_pending()); + } + + #[test] + pub fn test_taddr_stage_empty_range_clears() { + let tmp = TempDir::new("lctest").unwrap(); + let lc = LightClient::new(&main_config(&tmp), 0).unwrap(); + + lc.taddr_note_unscanned(500, &bt_from(&[(500, 1)])); + let called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let called_c = called.clone(); + // pass_start 11 -> t_start 11 > end 10: nothing to scan, the closure must not run and the debt clears. + let r = lc.taddress_stage(11, 10, &bt_from(&[]), move |_s, _e, _bt| { + called_c.store(true, std::sync::atomic::Ordering::SeqCst); + Ok(()) + }); + assert_eq!(r, Ok(())); + assert!(!called.load(std::sync::atomic::Ordering::SeqCst), "scan must not run for an empty range"); + assert!(!lc.taddr_has_pending()); + } + + #[test] + pub fn test_clear_state_clears_taddr_debt() { + let tmp = TempDir::new("lctest").unwrap(); + let lc = LightClient::new(&main_config(&tmp), 0).unwrap(); + + lc.taddr_note_unscanned(100, &bt_from(&[(100, 1)])); + assert!(lc.taddr_has_pending()); + lc.clear_state(); + assert!(!lc.taddr_has_pending(), "a rescan / clear_state cancels the owed t-scan"); + } + // C254 tests (s1 tests 9-14). Every mode is set explicitly before it is checked, never left to the umask. #[cfg(unix)] fn mode(p: &std::path::Path) -> u32 {