DragonX/ObsidianDragon
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

fix(lite-backend): fix t-scan rewind underflow and retry the owed scan

Browse Source 
When a sync pass's transparent (t-address) scan failed, the backend
mishandled it three ways: the retry computed its start height as the
block height minus the batch size, an unsigned subtraction that wrapped
to a huge value for low heights; the block range left un-scanned was
not remembered, so a transient failure left a permanent hole in the
transparent history that nothing ever revisited; and the block times
needed to date the transactions in that range were discarded.

- The owed range and the block times seen so far are now kept and folded
  into the next pass, which resumes from the lower of its own first block
  and any owed range.
- The scan start never subtracts, so it can no longer underflow.
- A t-address scan owed by an earlier failed pass is paid on the next
  sync even when no new block is pending.

Behaviour change: a transparent scan that fails is retried in-process on
the next sync rather than being lost. The owed range lives only in
memory and is never written to the wallet file, so the file format is
unchanged and remains interchangeable with other builds. The debt is
lost if the process exits or the server is switched before a retry
completes; Re-download blocks (rescan) recovers it, as today.

Tests: library tests on the extracted t-scan seam cover a failed pass
keeping its range and block times, a later success clearing the debt,
and the scan start never exceeding the pass start across the batch-size
boundary (the old formula wrapped there).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
DanS
2026-10-12 21:42:22 -05:00
co-authored by Claude Opus 4.8
parent 5ae5e5213b
commit 3f7af2d028
1 changed files with 241 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs
Vendored
+241 -18
View File
@@ -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<u64, u32>,
}
pub struct LightClient {
pub wallet : Arc<RwLock<LightWallet>>,
@@ -426,6 +434,7 @@ pub struct LightClient {
pub sapling_spend : Cow<'static, [u8]>,
sync_lock : Mutex<()>,
t_pending : Mutex<Option<TaddrPending>>, // leaf lock: never held while taking sync_lock or the wallet lock
sync_status : Arc<RwLock<WalletStatus>>, // The current syncing status of the Wallet.
pub shutdown_flag : Arc<AtomicBool>, // 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<LightClient>) -> 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<LightClient>) -> 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<LightClient>) -> Result<(), String> {
result.map(|(txid, _)| txid)
}
fn do_sync_internal(&self, print_updates: bool, retry_count: u32) -> Result<JsonValue, String> {
fn do_sync_internal(&self, print_updates: bool) -> Result<JsonValue, String> {
// 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<LightClient>) -> 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<RwLock<HashMap<u64, u32>>> = 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<LightClient>) -> 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<LightClient>) -> 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<LightClient>) -> 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<LightClient>) -> 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<LightClient>) -> 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<LightClient>) -> 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<RwLock<HashMap<u64, u32>>>) -> 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<RwLock<HashMap<u64, u32>>>) {
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<F>(&self, pass_start: u64, end_height: u64, block_times: &Arc<RwLock<HashMap<u64, u32>>>, scan: F)
-> Result<(), String>
where F: FnOnce(u64, u64, Arc<RwLock<HashMap<u64, u32>>>) -> 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<std::sync::RwLock<std::collections::HashMap<u64, u32>>>;
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 {