diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/fulltx_fetch.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/fulltx_fetch.rs new file mode 100644 index 0000000..3d4dec3 --- /dev/null +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/fulltx_fetch.rs @@ -0,0 +1,193 @@ +// Copyright The Hush Developers 2019-2022 +// Released under the GPLv3 + +// Bounded-concurrency fetch + scan of full transactions at the end of a sync +// (LightClient::scan_fill_fulltxs). One shared gRPC channel on the caller's Tokio runtime +// replaces a fresh runtime + TLS connection per txid; parse + scan stays on the sync ThreadPool. +// Privacy: which items are requested, in what order, how often they are retried and when the +// batch stops never depend on whether an item is one of the wallet's own txs. Only the final +// Ok/Err does. + +use std::collections::{BTreeSet, VecDeque}; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; + +use log::{info, warn}; +use threadpool::ThreadPool; +use tokio::runtime::Runtime; +use zcash_primitives::transaction::{Transaction, TxId}; + +use crate::lightwallet::LightWallet; + +/// One GetTransaction attempt; resolves to the raw transaction bytes. +pub type FetchFuture = Pin, String>> + Send>>; +/// fetch(txid, reconnect): reconnect == true means "dial a new shared connection first". +pub type FetchFn = Arc FetchFuture + Send + Sync>; +/// scan(raw_tx_bytes, height): parse + LightWallet::scan_full_tx. Always runs on the ThreadPool. +pub type ScanFn = Arc, i32) -> Result<(), String> + Send + Sync>; + +#[derive(Clone, Copy, Debug)] +pub struct FetchPolicy { + pub workers: usize, + pub max_attempts: u32, + pub backoff_ms: u64, + pub attempt_timeout: Duration, + pub fail_breaker: usize, +} + +impl FetchPolicy { + /// Production policy. Fetch concurrency == the sync ThreadPool size (max(2, min(8, cpus))), + /// i.e. exactly as many concurrent GetTransaction calls as the old one-job-per-txid code. + pub fn for_pool(pool: &ThreadPool) -> FetchPolicy { + FetchPolicy { + workers: pool.max_count(), + max_attempts: 3, + backoff_ms: 250, + attempt_timeout: Duration::from_secs(120), + fail_breaker: 16, + } + } +} + +/// Exactly the old list: own ∪ decoys, sorted by (txid, height), exact duplicates removed. +/// Returns (list, own_set); an item is "own" iff its (txid, height) is in own_set. +pub fn plan_fetch_list(own: Vec<(TxId, i32)>, decoys: Vec<(TxId, i32)>) + -> (Vec<(TxId, i32)>, BTreeSet<(TxId, i32)>) +{ + let own_set: BTreeSet<(TxId, i32)> = own.iter().cloned().collect(); + let mut all = own; + all.extend_from_slice(&decoys[..]); + all.sort(); + all.dedup(); + (all, own_set) +} + +/// The production scan step: identical wallet effect to the old pool job +/// (`scan_full_tx(&tx, height, 0)`), but unparsable bytes are an Err instead of a panic. +pub fn wallet_scan_fn(wallet: Arc>) -> ScanFn { + Arc::new(move |bytes: Vec, height: i32| { + let tx = match Transaction::read(&bytes[..]) { + Ok(tx) => tx, + Err(e) => return Err(format!("Error reading transaction: {:?}", e)), + }; + wallet.read().unwrap().scan_full_tx(&tx, height, 0); + Ok(()) + }) +} + +/// Fetch + scan every item. Ok iff every own item was fetched and scanned. +/// Must be called from a thread that is not inside a Tokio runtime (the sync thread). +pub fn run_fetch_scan( + rt: &mut Runtime, + items: Vec<(TxId, i32)>, + own_set: BTreeSet<(TxId, i32)>, + policy: FetchPolicy, + pool: ThreadPool, + fetch: FetchFn, + scan: ScanFn, +) -> Result<(), String> { + let own_total = items.iter().filter(|item| own_set.contains(*item)).count(); + let total = items.len(); + let workers = std::cmp::max(1, std::cmp::min(policy.workers, total)); + let queue = Arc::new(Mutex::new(items.into_iter().collect::>())); + let own_set = Arc::new(own_set); + let own_ok = Arc::new(AtomicUsize::new(0)); + let streak = Arc::new(AtomicUsize::new(0)); + let abort = Arc::new(AtomicBool::new(false)); + let first_err: Arc>> = Arc::new(Mutex::new(None)); + + rt.block_on(async { + let mut handles = Vec::with_capacity(workers); + for _ in 0..workers { + let queue = queue.clone(); + let own_set = own_set.clone(); + let own_ok = own_ok.clone(); + let streak = streak.clone(); + let abort = abort.clone(); + let first_err = first_err.clone(); + let pool = pool.clone(); + let fetch = fetch.clone(); + let scan = scan.clone(); + handles.push(tokio::spawn(async move { + loop { + if abort.load(Ordering::SeqCst) { + break; + } + let next = queue.lock().unwrap().pop_front(); + let (txid, height) = match next { + Some(item) => item, + None => break, + }; + let mut result: Result<(), String> = Err("not attempted".to_string()); + for attempt in 0..policy.max_attempts { + if attempt > 0 { + tokio::time::delay_for(Duration::from_millis(policy.backoff_ms << (attempt - 1))).await; + } + let fetched = match tokio::time::timeout(policy.attempt_timeout, fetch(txid, attempt > 0)).await { + Ok(r) => r, + Err(_) => Err(format!("Error in get_transaction runtime timed out after {:?}", policy.attempt_timeout)), + }; + let bytes = match fetched { + Ok(bytes) => bytes, + Err(e) => { + result = Err(e); + continue; + } + }; + // Parse + scan on the pool. A panic there drops `done`, which turns into + // Err here instead of a completion that never arrives (the old wedge). + let (done, finished) = tokio::sync::oneshot::channel::>(); + let scan = scan.clone(); + pool.execute(move || { + let _ = done.send(scan(bytes, height)); + }); + result = match finished.await { + Ok(r) => r, + Err(_) => Err("full tx scan job panicked".to_string()), + }; + break; // parse/scan failures are deterministic: never re-fetch + } + match result { + Ok(()) => { + streak.store(0, Ordering::SeqCst); + if own_set.contains(&(txid, height)) { + own_ok.fetch_add(1, Ordering::SeqCst); + } + } + Err(e) => { + warn!("Could not fetch full tx {}: {}", txid, e); + if own_set.contains(&(txid, height)) { + let mut first = first_err.lock().unwrap(); + if first.is_none() { + *first = Some(e); + } + } + // Ownership-blind breaker: a dead network stops the batch; a failed own + // tx alone never does (stopping on it would tell the server it is ours). + if streak.fetch_add(1, Ordering::SeqCst) + 1 >= policy.fail_breaker { + abort.store(true, Ordering::SeqCst); + } + } + } + } + })); + } + for handle in handles { + let _ = handle.await; // a panicked worker shows up below as own_ok < own_total + } + }); + + let ok = own_ok.load(Ordering::SeqCst); + if ok < own_total { + let first = match first_err.lock() { + Ok(mut guard) => guard.take(), + Err(poisoned) => poisoned.into_inner().take(), + }; + return Err(first.unwrap_or_else(|| format!("fetched only {} of {} wallet txs", ok, own_total))); + } + info!("Full tx fetch done: {} items, {} new", total, own_total); + Ok(()) +} diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/grpcconnector.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/grpcconnector.rs index ba08292..79c2401 100644 --- a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/grpcconnector.rs +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/grpcconnector.rs @@ -307,6 +307,49 @@ pub fn fetch_full_tx(uri: &http::Uri, txid: TxId, no_cert: bool) -> Result crate::fulltx_fetch::FetchFn { + let slot: Arc>>> = + Arc::new(std::sync::Mutex::new(None)); + Arc::new(move |txid: TxId, reconnect: bool| { + let slot = slot.clone(); + let uri = uri.clone(); + Box::pin(async move { + let cached = if reconnect { + None + } else { + match slot.lock() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } + }; + let mut client = match cached { + Some(c) => c, + None => { + let c = match get_client(&uri, no_cert).await { + Ok(c) => c, + Err(e) => return Err(format!("Error in get_transaction runtime {}", e.to_string())), + }; + match slot.lock() { + Ok(mut guard) => *guard = Some(c.clone()), + Err(poisoned) => *poisoned.into_inner() = Some(c.clone()), + } + c + } + }; + let request = Request::new(TxFilter { block: None, index: 0, hash: txid.0.to_vec() }); + match client.get_transaction(request).await { + Ok(response) => Ok(response.into_inner().data.to_vec()), + Err(e) => Err(format!("Error in get_transaction runtime {}", e.to_string())), + } + }) as crate::fulltx_fetch::FetchFuture + }) +} + // send_transaction GRPC call async fn send_transaction(uri: &http::Uri, no_cert: bool, tx_bytes: Box<[u8]>) -> Result> { let mut client = get_client(uri, no_cert).await?; diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lib.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lib.rs index 4f6ad39..b46e5cc 100644 --- a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lib.rs +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lib.rs @@ -5,6 +5,7 @@ extern crate rust_embed; pub mod lightclient; pub mod grpcconnector; +pub mod fulltx_fetch; pub mod lightwallet; pub mod commands; diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs index a020f1e..ca973ae 100644 --- a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightclient.rs @@ -1590,7 +1590,7 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { { let decoy_txids = all_new_txs.read().unwrap(); - match self.scan_fill_fulltxs(&pool, decoy_txids.to_vec()) { + match self.scan_fill_fulltxs(&mut rt, &pool, decoy_txids.to_vec()) { Ok(_) => Ok(object!{ "result" => "success", "latest_block" => latest_block, @@ -1677,50 +1677,26 @@ pub fn start_mempool_monitor(lc: Arc) -> Result<(), String> { } - fn scan_fill_fulltxs(&self, pool: &ThreadPool, decoy_txids: Vec<(TxId, i32)>) -> Result, String> { + fn scan_fill_fulltxs(&self, rt: &mut Runtime, pool: &ThreadPool, decoy_txids: Vec<(TxId, i32)>) -> Result<(), String> { // We need to first copy over the Txids from the wallet struct, because // we need to free the read lock from here (Because we'll self.wallet.txs later) - let mut txids_to_fetch: Vec<(TxId, i32)> = self.wallet.read().unwrap().txs.read().unwrap().values() + let own_txids: Vec<(TxId, i32)> = self.wallet.read().unwrap().txs.read().unwrap().values() .filter(|wtx| wtx.full_tx_scanned == false) .map(|wtx| (wtx.txid.clone(), wtx.block)) .collect::>(); - info!("Fetching {} new txids, total {} with decoy", txids_to_fetch.len(), decoy_txids.len()); - txids_to_fetch.extend_from_slice(&decoy_txids[..]); - txids_to_fetch.sort(); - txids_to_fetch.dedup(); + info!("Fetching {} new txids, total {} with decoy", own_txids.len(), decoy_txids.len()); - let num_fetches = txids_to_fetch.len(); - let (ctx, crx) = channel(); + // Own txs and decoys go out as ONE (txid, height)-sorted, de-duplicated list, exactly as + // before, over one shared connection on the sync runtime. Request order, retries and the + // stop rule are identical for own txs and decoys, so the server cannot tell them apart. + let (items, own_set) = crate::fulltx_fetch::plan_fetch_list(own_txids, decoy_txids); + let policy = crate::fulltx_fetch::FetchPolicy::for_pool(pool); + let fetch = grpcconnector::make_shared_tx_fetcher(self.get_server_uri(), self.config.no_cert_verification); + let scan = crate::fulltx_fetch::wallet_scan_fn(self.wallet.clone()); - // And go and fetch the txids, getting the full transaction, so we can - // read the memos - for (txid, height) in txids_to_fetch { - let light_wallet_clone = self.wallet.clone(); - - let pool = pool.clone(); - let server_uri = self.get_server_uri(); - let ctx = ctx.clone(); - let no_cert = self.config.no_cert_verification; - - pool.execute(move || { - //info!("Fetching full Tx: {}", txid); - - match fetch_full_tx(&server_uri, txid, no_cert) { - Ok(tx_bytes) => { - let tx = Transaction::read(&tx_bytes[..]).unwrap(); - - light_wallet_clone.read().unwrap().scan_full_tx(&tx, height, 0); - ctx.send(Ok(())).unwrap(); - }, - Err(e) => ctx.send(Err(e)).unwrap() - }; - }); - }; - - // Wait for all the fetches to finish. - crx.iter().take(num_fetches).collect::, String>>() + crate::fulltx_fetch::run_fetch_scan(rt, items, own_set, policy, pool.clone(), fetch, scan) } pub fn do_send(&self, addrs: Vec<(&str, u64, Option)>, fee: &u64) -> Result { diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/tests/fulltx_fetch.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/tests/fulltx_fetch.rs new file mode 100644 index 0000000..c4f9724 --- /dev/null +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/tests/fulltx_fetch.rs @@ -0,0 +1,361 @@ +// Offline tests for the full-tx fetch engine (src/fulltx_fetch.rs). No network, no proving. +// Integration test on purpose: the in-crate #[cfg(test)] modules do not compile on BASE. +use std::collections::{BTreeSet, HashMap}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc, Mutex, RwLock}; +use std::time::{Duration, Instant}; + +use base58::FromBase58; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use threadpool::ThreadPool; +use tokio::runtime::Runtime; +use zcash_primitives::legacy::{Script, TransparentAddress}; +use zcash_primitives::transaction::components::{Amount, OutPoint, TxIn, TxOut}; +use zcash_primitives::transaction::{Transaction, TransactionData, TxId}; + +use silentdragonxlitelib::fulltx_fetch::{ + plan_fetch_list, run_fetch_scan, wallet_scan_fn, FetchFn, FetchFuture, FetchPolicy, ScanFn, +}; +use silentdragonxlitelib::lightclient::LightClientConfig; +use silentdragonxlitelib::lightwallet::LightWallet; + +type CallLog = Arc>>; +type ScanLog = Arc>>; + +fn tid(n: u32) -> TxId { + let mut b = [0u8; 32]; + b[..4].copy_from_slice(&n.to_be_bytes()); // big-endian prefix => TxId order == n order + TxId(b) +} + +fn policy(workers: usize) -> FetchPolicy { + FetchPolicy { workers, max_attempts: 3, backoff_ms: 1, attempt_timeout: Duration::from_secs(10), fail_breaker: 16 } +} + +fn run(items: Vec<(TxId, i32)>, own: &[(TxId, i32)], p: FetchPolicy, fetch: FetchFn, scan: ScanFn) -> Result<(), String> { + let own_set: BTreeSet<(TxId, i32)> = own.iter().cloned().collect(); + let mut rt = Runtime::new().unwrap(); + run_fetch_scan(&mut rt, items, own_set, p, ThreadPool::new(8), fetch, scan) +} + +/// Fake GetTransaction: logs (txid, reconnect); returns the txid bytes, or Err when +/// `fail(txid, nth_call_for_this_txid)` says so. +fn fake_fetch(log: CallLog, delay_ms: u64, fail: F) -> FetchFn +where + F: Fn(TxId, usize) -> bool + Send + Sync + 'static, +{ + let calls: Arc>> = Arc::new(Mutex::new(HashMap::new())); + let fail = Arc::new(fail); + Arc::new(move |txid: TxId, reconnect: bool| { + let log = log.clone(); + let calls = calls.clone(); + let fail = fail.clone(); + Box::pin(async move { + log.lock().unwrap().push((txid, reconnect)); + let nth = { + let mut c = calls.lock().unwrap(); + let e = c.entry(txid).or_insert(0); + *e += 1; + *e + }; + if delay_ms > 0 { + tokio::time::delay_for(Duration::from_millis(delay_ms)).await; + } + if (*fail)(txid, nth) { Err(format!("fake failure {}", txid)) } else { Ok(txid.0.to_vec()) } + }) as FetchFuture + }) +} + +fn logging_scan(log: ScanLog) -> ScanFn { + Arc::new(move |bytes: Vec, height: i32| { + let mut b = [0u8; 32]; + b.copy_from_slice(&bytes[..32]); + log.lock().unwrap().push((TxId(b), height)); + Ok(()) + }) +} + +fn no_scan_log() -> ScanFn { logging_scan(Arc::new(Mutex::new(vec![]))) } + +#[test] // T1 +fn plan_is_exactly_the_old_list() { + let mut rng = StdRng::seed_from_u64(7); + for round in 0..300 { + let n_own = rng.gen_range(0, 12); + let own: Vec<(TxId, i32)> = (0..n_own).map(|_| (tid(rng.gen_range(0, 40)), rng.gen_range(1, 4))).collect(); + let n_dec = rng.gen_range(0, 60); + let decoys: Vec<(TxId, i32)> = (0..n_dec).map(|_| (tid(rng.gen_range(0, 40)), rng.gen_range(1, 4))).collect(); + let mut old = own.clone(); + old.extend_from_slice(&decoys[..]); + old.sort(); + old.dedup(); + let (items, own_set) = plan_fetch_list(own.clone(), decoys); + assert_eq!(items, old, "round {}", round); + assert_eq!(own_set, own.iter().cloned().collect::>()); + } +} + +#[test] // T2 +fn every_item_fetched_and_scanned_exactly_once() { + for &n in &[0u32, 1, 7, 200] { + let items: Vec<(TxId, i32)> = (0..n).map(|i| (tid(i), (i % 5) as i32 + 1)).collect(); + let own: Vec<(TxId, i32)> = items.iter().cloned().step_by(3).collect(); + let calls: CallLog = Arc::new(Mutex::new(vec![])); + let scans: ScanLog = Arc::new(Mutex::new(vec![])); + let r = run(items.clone(), &own, policy(8), fake_fetch(calls.clone(), 1, |_, _| false), logging_scan(scans.clone())); + assert_eq!(r, Ok(())); + let mut fetched: Vec = calls.lock().unwrap().iter().map(|(t, re)| { assert!(!*re); *t }).collect(); + fetched.sort(); + let mut scanned = scans.lock().unwrap().clone(); + scanned.sort(); + assert_eq!(fetched, items.iter().map(|(t, _)| *t).collect::>()); + assert_eq!(scanned, items); + } +} + +#[test] // T3 +fn concurrency_is_bounded_by_workers() { + for &workers in &[1usize, 2, 8] { + let inflight = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let fetch: FetchFn = { + let (inflight, peak) = (inflight.clone(), peak.clone()); + Arc::new(move |txid: TxId, _r: bool| { + let (inflight, peak) = (inflight.clone(), peak.clone()); + Box::pin(async move { + let now = inflight.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(now, Ordering::SeqCst); + tokio::time::delay_for(Duration::from_millis(5)).await; + inflight.fetch_sub(1, Ordering::SeqCst); + Ok::, String>(txid.0.to_vec()) + }) as FetchFuture + }) + }; + let items: Vec<(TxId, i32)> = (0..200).map(|i| (tid(i), 1)).collect(); + assert_eq!(run(items, &[], policy(workers), fetch, no_scan_log()), Ok(())); + let p = peak.load(Ordering::SeqCst); + assert!(p <= workers, "peak {} > workers {}", p, workers); + if workers > 1 { assert!(p >= 2, "fetches never overlapped"); } + } +} + +#[test] // T4: privacy guard +fn request_sequence_does_not_depend_on_ownership() { + let items: Vec<(TxId, i32)> = (0..40).map(|i| (tid(i), 7)).collect(); + let evens: Vec<(TxId, i32)> = items.iter().cloned().filter(|(t, _)| t.0[3] % 2 == 0).collect(); + let first10: Vec<(TxId, i32)> = items[..10].to_vec(); + let mut logs = vec![]; + for own in &[evens, first10, vec![]] { + let calls: CallLog = Arc::new(Mutex::new(vec![])); + let fetch = fake_fetch(calls.clone(), 0, |t, nth| t == tid(5) || t == tid(17) || (t == tid(9) && nth == 1)); + let _ = run(items.clone(), own, policy(1), fetch, no_scan_log()); + logs.push(calls.lock().unwrap().clone()); + } + assert_eq!(logs[0], logs[1]); + assert_eq!(logs[0], logs[2]); + let mut seen: Vec = vec![]; + for (t, _) in &logs[0] { if !seen.contains(t) { seen.push(*t); } } + assert_eq!(seen, items.iter().map(|(t, _)| *t).collect::>()); // FIFO, plan order +} + +#[test] // T5 +fn decoy_failures_are_ignored_and_retries_redial() { + let items: Vec<(TxId, i32)> = (0..10).map(|i| (tid(i), 3)).collect(); + let calls: CallLog = Arc::new(Mutex::new(vec![])); + let scans: ScanLog = Arc::new(Mutex::new(vec![])); + let fetch = fake_fetch(calls.clone(), 0, |t, nth| t == tid(5) || t == tid(7) || (t == tid(3) && nth == 1)); + assert_eq!(run(items, &[(tid(3), 3)], policy(4), fetch, logging_scan(scans.clone())), Ok(())); + let calls = calls.lock().unwrap().clone(); + let flags = |id: TxId| calls.iter().filter(|(t, _)| *t == id).map(|(_, r)| *r).collect::>(); + assert_eq!(flags(tid(3)), vec![false, true]); + assert_eq!(flags(tid(5)), vec![false, true, true]); + assert_eq!(flags(tid(0)), vec![false]); + assert!(scans.lock().unwrap().contains(&(tid(3), 3))); +} + +#[test] // T6 +fn own_failure_is_an_error_but_never_stops_the_batch() { + let items: Vec<(TxId, i32)> = (0..20).map(|i| (tid(i), 1)).collect(); + let calls: CallLog = Arc::new(Mutex::new(vec![])); + let scans: ScanLog = Arc::new(Mutex::new(vec![])); + let r = run(items, &[(tid(0), 1), (tid(10), 1)], policy(1), fake_fetch(calls.clone(), 0, |t, _| t == tid(0)), logging_scan(scans.clone())); + assert!(r.unwrap_err().contains("fake failure")); + let fetched: BTreeSet = calls.lock().unwrap().iter().map(|(t, _)| *t).collect(); + assert_eq!(fetched.len(), 20); + assert!(scans.lock().unwrap().contains(&(tid(10), 1))); +} + +#[test] // T7 +fn dead_network_trips_the_breaker() { + let p = policy(8); + for &with_own in &[true, false] { + let items: Vec<(TxId, i32)> = (0..1000).map(|i| (tid(i), 1)).collect(); + let own = if with_own { vec![(tid(500), 1), (tid(999), 1)] } else { vec![] }; + let calls: CallLog = Arc::new(Mutex::new(vec![])); + let start = Instant::now(); + let r = run(items, &own, p, fake_fetch(calls.clone(), 0, |_, _| true), no_scan_log()); + assert_eq!(r.is_err(), with_own); + let n = calls.lock().unwrap().len(); + assert!(n <= (p.fail_breaker + 2 * p.workers) * p.max_attempts as usize, "{} calls", n); + assert!(start.elapsed() < Duration::from_secs(20)); + } +} + +#[test] // T8 +fn stalled_request_times_out_and_redials() { + let calls: CallLog = Arc::new(Mutex::new(vec![])); + let fetch: FetchFn = { + let calls = calls.clone(); + Arc::new(move |txid: TxId, reconnect: bool| { + let calls = calls.clone(); + Box::pin(async move { + calls.lock().unwrap().push((txid, reconnect)); + if !reconnect { tokio::time::delay_for(Duration::from_secs(30)).await; } + Ok::, String>(txid.0.to_vec()) + }) as FetchFuture + }) + }; + let mut p = policy(2); + p.attempt_timeout = Duration::from_millis(100); + let start = Instant::now(); + assert_eq!(run(vec![(tid(1), 1), (tid(2), 1)], &[(tid(1), 1)], p, fetch, no_scan_log()), Ok(())); + assert!(start.elapsed() < Duration::from_secs(5)); + assert_eq!(calls.lock().unwrap().iter().filter(|(t, _)| *t == tid(1)).count(), 2); +} + +#[test] // T9 +fn panicking_scan_never_hangs() { + for &(own_panics, expect_ok) in &[(false, true), (true, false)] { + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let items: Vec<(TxId, i32)> = (0..30).map(|i| (tid(i), 1)).collect(); + let bad = if own_panics { tid(1) } else { tid(2) }; + let scan: ScanFn = Arc::new(move |bytes: Vec, _h: i32| { + if bytes[..] == bad.0[..] { panic!("scan boom"); } + Ok(()) + }); + let r = run(items, &[(tid(1), 1)], policy(4), fake_fetch(Arc::new(Mutex::new(vec![])), 0, |_, _| false), scan); + let _ = tx.send(r); + }); + let r = rx.recv_timeout(Duration::from_secs(30)).expect("engine hung after a scan panic"); + assert_eq!(r.is_ok(), expect_ok); + } +} + +fn main_config() -> LightClientConfig { + // chain "main": LightClientConfig::base58_pubkey_address() panics for any other chain name. + LightClientConfig { + server: "0.0.0.0:0".parse().unwrap(), + chain_name: "main".to_string(), + sapling_activation_height: 0, + consensus_branch_id: "000000".to_string(), + anchor_offset: 0, + no_cert_verification: false, + data_dir: None, + } +} + +#[test] // T10 +fn unparsable_tx_bytes_are_an_error_not_a_panic_or_hang() { + let wallet = Arc::new(RwLock::new(LightWallet::new(None, &main_config(), 0, 0).unwrap())); + let scan = wallet_scan_fn(wallet); + assert!(scan(vec![], 1).is_err()); + for &(own_bad, expect_ok) in &[(true, false), (false, true)] { + let fetch: FetchFn = Arc::new(move |_txid: TxId, _r: bool| { + Box::pin(async move { Ok::, String>(vec![0xffu8; 64]) }) as FetchFuture + }); + let own = if own_bad { vec![(tid(1), 1)] } else { vec![] }; + let scan = scan.clone(); + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(run(vec![(tid(1), 1), (tid(2), 1)], &own, policy(2), fetch, scan)); + }); + let r = rx.recv_timeout(Duration::from_secs(30)).expect("engine hung on garbage bytes"); + assert_eq!(r.is_ok(), expect_ok); + } +} + +fn my_taddr_hash(wallet: &LightWallet) -> [u8; 20] { + let raw = wallet.get_all_taddresses()[0].from_base58().unwrap(); + assert_eq!(raw.len(), 25, "prefix(1) + hash160(20) + checksum(4)"); + let mut h = [0u8; 20]; + h.copy_from_slice(&raw[1..21]); + h +} + +fn fake_tx(txid: TxId, spends: &[(TxId, u32)], pays: &[([u8; 20], u64)]) -> Transaction { + let mut tx = Transaction { txid, data: TransactionData::new() }; + for &(prev, n) in spends { + tx.data.vin.push(TxIn { prevout: OutPoint { hash: prev.0, n }, script_sig: Script(vec![]), sequence: 0 }); + } + for &(hash, value) in pays { + tx.data.vout.push(TxOut { value: Amount::from_u64(value).unwrap(), script_pubkey: TransparentAddress::PublicKey(hash).script() }); + } + tx +} + +fn wallet_bytes(w: &Arc>) -> Vec { + let mut v = vec![]; + w.read().unwrap().write(&mut v).unwrap(); + v +} + +#[test] // T11: the engine leaves the same wallet file as the old serial fill +fn engine_leaves_same_wallet_bytes_as_serial_fill() { + let cfg = main_config(); + let w0 = LightWallet::new(None, &cfg, 0, 0).unwrap(); + let me = my_taddr_hash(&w0); + let other = [7u8; 20]; + let mut txs: HashMap = HashMap::new(); + // 12 own txs paying us. The first one expands the HD gap (serially, before the snapshot), so the + // batch below never races ensure_hd_taddresses (a pre-existing race that exists today too). + let mut own: Vec<(TxId, i32)> = vec![]; + for i in 0..12u32 { + let tx = fake_tx(tid(1000 + i), &[], &[(me, 1000 + i as u64), (other, 5)]); + w0.scan_full_tx(&tx, 10 + i as i32, 0); + own.push((tx.txid, 10 + i as i32)); + txs.insert(tx.txid, tx); + } + assert!(w0.tbalance(None) > 0, "t-address hash did not match the wallet"); + for (id, _) in &own { w0.txs.write().unwrap().get_mut(id).unwrap().full_tx_scanned = false; } + let mut snap = vec![]; + w0.write(&mut snap).unwrap(); + // Decoys with no intra-batch dependencies: spenders of pre-batch utxos, payers to us, unrelated. + let mut decoys: Vec<(TxId, i32)> = own.clone(); + for i in 0..6u32 { let t = fake_tx(tid(2000 + i), &[(tid(1000 + i), 0)], &[(other, 900)]); decoys.push((t.txid, 200 + i as i32)); txs.insert(t.txid, t); } + for i in 0..8u32 { let t = fake_tx(tid(3000 + i), &[], &[(me, 77 + i as u64)]); decoys.push((t.txid, 300 + i as i32)); txs.insert(t.txid, t); } + for i in 0..30u32 { let t = fake_tx(tid(4000 + i), &[], &[(other, 1)]); decoys.push((t.txid, 400 + i as i32)); txs.insert(t.txid, t); } + decoys.push((tid(4000), 999)); // same unrelated txid at a second height + let (items, own_set) = plan_fetch_list(own.clone(), decoys); + let txs = Arc::new(txs); + + let wa = Arc::new(RwLock::new(LightWallet::read(&snap[..], &cfg).unwrap())); + for (id, h) in &items { wa.read().unwrap().scan_full_tx(&txs[id], *h, 0); } // old fill, serial + let expected = wallet_bytes(&wa); + + for seed in 0..5u8 { + let wb = Arc::new(RwLock::new(LightWallet::read(&snap[..], &cfg).unwrap())); + let scan: ScanFn = { + let (wb, txs) = (wb.clone(), txs.clone()); + Arc::new(move |bytes: Vec, h: i32| { + let mut b = [0u8; 32]; + b.copy_from_slice(&bytes[..32]); + match txs.get(&TxId(b)) { + Some(tx) => { wb.read().unwrap().scan_full_tx(tx, h, 0); Ok(()) } + None => Err("unknown txid".to_string()), + } + }) + }; + let fetch: FetchFn = Arc::new(move |txid: TxId, _r: bool| { + Box::pin(async move { + tokio::time::delay_for(Duration::from_millis(((txid.0[3] ^ seed) % 4) as u64)).await; + Ok::, String>(txid.0.to_vec()) + }) as FetchFuture + }); + let mut rt = Runtime::new().unwrap(); + assert_eq!(run_fetch_scan(&mut rt, items.clone(), own_set.clone(), policy(8), ThreadPool::new(8), fetch, scan), Ok(())); + assert!(wallet_bytes(&wb) == expected, "wallet bytes differ from the serial fill (seed {})", seed); + for (id, _) in &own { assert!(wb.read().unwrap().txs.read().unwrap()[id].full_tx_scanned); } + } +}