perf(lite-backend): fetch full transactions over one shared gRPC channel
At the end of a sync the wallet fetches the full transaction of each new wallet tx, together with every other transaction in the blocks that contain them, as decoys. Every txid got its own pool job that built a fresh Tokio runtime (one thread per CPU), a new rustls config and a new TCP + TLS + HTTP/2 connection for a single GetTransaction call, so a restore or rescan opened one connection per fetched transaction. The first error returned from the sync while the remaining jobs kept fetching and scanning in the background, and a stalled connection, an unparsable transaction or a panicking scan could hang the sync forever with its lock held. fulltx_fetch.rs now works through the same (txid, height)-sorted, de-duplicated list on the sync's runtime, with as many workers as the sync pool has threads (2-8, the same fan-out as before), over one gRPC channel that is dialed lazily and re-dialed on retry. Parsing and scanning stay on the sync pool, and the call returns once the whole batch has finished. Behaviour changes: - a failure on a decoy no longer fails the sync; it fails only when one of the wallet's own txs could not be fetched and scanned; - new error texts "Error reading transaction: ...", "full tx scan job panicked" and "fetched only X of Y wallet txs", all after the existing "Error fetching all txns for memos: " prefix; - each tx gets up to 3 fetch attempts with a 120 s timeout each, where the old code could wait forever; a request that legitimately takes longer now fails; - nothing keeps running in the background after the sync returns; - all GetTransaction calls share one TLS/HTTP-2 connection (up to one dial per worker at the start, plus a re-dial per retry); the request list and its order are unchanged; - if the network silently drops packets mid-batch, every attempt waits out its timeout until 16 consecutive failed txs stop the batch: up to about 12 minutes with 8 workers (48 minutes with 2) with the sync lock held, where the old code gave up at its first failed connection. Which txs are requested, in what order, how often they are retried and when the batch stops do not depend on which txs are the wallet's own; only the final result does. A server that keeps failing one tx can see that result through the sync's retry, but it could already tell, because an own tx that failed is requested again on the next sync without its decoys; nothing new leaks. tests/fulltx_fetch.rs drives the engine with a fake fetcher: the old request list, each item fetched and scanned once, bounded concurrency, a request sequence independent of ownership, retries, timeouts, the breaker, panicking and unparsable scans, and a saved wallet byte-identical to the old serial fill. The gRPC fetcher itself is not covered by these tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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<Box<dyn Future<Output = Result<Vec<u8>, String>> + Send>>;
|
||||
/// fetch(txid, reconnect): reconnect == true means "dial a new shared connection first".
|
||||
pub type FetchFn = Arc<dyn Fn(TxId, bool) -> FetchFuture + Send + Sync>;
|
||||
/// scan(raw_tx_bytes, height): parse + LightWallet::scan_full_tx. Always runs on the ThreadPool.
|
||||
pub type ScanFn = Arc<dyn Fn(Vec<u8>, 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<RwLock<LightWallet>>) -> ScanFn {
|
||||
Arc::new(move |bytes: Vec<u8>, 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::<VecDeque<(TxId, i32)>>()));
|
||||
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<Mutex<Option<String>>> = 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::<Result<(), String>>();
|
||||
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(())
|
||||
}
|
||||
@@ -307,6 +307,49 @@ pub fn fetch_full_tx(uri: &http::Uri, txid: TxId, no_cert: bool) -> Result<Vec<u
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared-connection fetcher for fulltx_fetch::run_fetch_scan. Every concurrent GetTransaction is
|
||||
/// multiplexed over ONE tonic Channel (cheap Clone; one TLS + HTTP/2 connection) instead of a new
|
||||
/// Tokio runtime + TLS connection per txid (fetch_full_tx). The channel is dialed lazily on first
|
||||
/// use, inside the caller's runtime (Channel spawns its buffer task there), and re-dialed when the
|
||||
/// engine retries with reconnect == true. Error text matches fetch_full_tx.
|
||||
pub fn make_shared_tx_fetcher(uri: http::Uri, no_cert: bool) -> crate::fulltx_fetch::FetchFn {
|
||||
let slot: Arc<std::sync::Mutex<Option<CompactTxStreamerClient<Channel>>>> =
|
||||
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<String, Box<dyn std::error::Error>> {
|
||||
let mut client = get_client(uri, no_cert).await?;
|
||||
|
||||
@@ -5,6 +5,7 @@ extern crate rust_embed;
|
||||
|
||||
pub mod lightclient;
|
||||
pub mod grpcconnector;
|
||||
pub mod fulltx_fetch;
|
||||
pub mod lightwallet;
|
||||
pub mod commands;
|
||||
|
||||
|
||||
+12
-36
@@ -1590,7 +1590,7 @@ pub fn start_mempool_monitor(lc: Arc<LightClient>) -> 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<LightClient>) -> Result<(), String> {
|
||||
}
|
||||
|
||||
|
||||
fn scan_fill_fulltxs(&self, pool: &ThreadPool, decoy_txids: Vec<(TxId, i32)>) -> Result<Vec<()>, 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::<Vec<(TxId, i32)>>();
|
||||
|
||||
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::<Result<Vec<()>, 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<String>)>, fee: &u64) -> Result<String, String> {
|
||||
|
||||
+361
@@ -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<Mutex<Vec<(TxId, bool)>>>;
|
||||
type ScanLog = Arc<Mutex<Vec<(TxId, i32)>>>;
|
||||
|
||||
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<F>(log: CallLog, delay_ms: u64, fail: F) -> FetchFn
|
||||
where
|
||||
F: Fn(TxId, usize) -> bool + Send + Sync + 'static,
|
||||
{
|
||||
let calls: Arc<Mutex<HashMap<TxId, usize>>> = 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<u8>, 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::<BTreeSet<_>>());
|
||||
}
|
||||
}
|
||||
|
||||
#[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<TxId> = 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::<Vec<TxId>>());
|
||||
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::<Vec<u8>, 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<TxId> = vec![];
|
||||
for (t, _) in &logs[0] { if !seen.contains(t) { seen.push(*t); } }
|
||||
assert_eq!(seen, items.iter().map(|(t, _)| *t).collect::<Vec<TxId>>()); // 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::<Vec<bool>>();
|
||||
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<TxId> = 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::<Vec<u8>, 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<u8>, _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::<Vec<u8>, 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<RwLock<LightWallet>>) -> Vec<u8> {
|
||||
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<TxId, Transaction> = 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<u8>, 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::<Vec<u8>, 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); }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user