Merge branch 'danger'

Cargo.lock

@@ -174,7 +174,7 @@ dependencies = [
  "futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures-cpupool 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "group 0.1.0 (git+https://github.com/MyHush/librustzcash.git?rev=1a0204113d487cdaaf183c2967010e5214ff9e37)",
- "num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.13.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "pairing 0.14.2 (git+https://github.com/MyHush/librustzcash.git?rev=1a0204113d487cdaaf183c2967010e5214ff9e37)",
  "rand_core 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -559,7 +559,7 @@ version = "0.1.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)",
- "num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.13.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1000,7 +1000,7 @@ dependencies = [
 
 [[package]]
 name = "num_cpus"
-version = "1.11.1"
+version = "1.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "hermit-abi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1755,6 +1755,7 @@ dependencies = [
  "libflate 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "log4rs 0.8.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.13.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "pairing 0.14.2 (git+https://github.com/MyHush/librustzcash.git?rev=1a0204113d487cdaaf183c2967010e5214ff9e37)",
  "prost 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "prost-types 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1766,7 +1767,9 @@ dependencies = [
  "secp256k1 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "sha2 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "sodiumoxide 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "subtle 2.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "tempdir 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "threadpool 1.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tiny-bip39 0.6.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-rustls 0.12.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1933,6 +1936,14 @@ dependencies = [
  "lazy_static 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "threadpool"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "num_cpus 1.13.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "time"
 version = "0.1.42"
@@ -1972,7 +1983,7 @@ dependencies = [
  "mio 0.6.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio-named-pipes 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio-uds 0.6.7 (registry+https://github.com/rust-lang/crates.io-index)",
- "num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.13.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "signal-hook-registry 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -2707,7 +2718,7 @@ dependencies = [
 "checksum num-bigint 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "f9c3f34cdd24f334cb265d9bf8bfa8a241920d026916785747a92f0e55541a1a"
 "checksum num-integer 0.1.41 (registry+https://github.com/rust-lang/crates.io-index)" = "b85e541ef8255f6cf42bbfe4ef361305c6c135d10919ecc26126c4e5ae94bc09"
 "checksum num-traits 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)" = "d4c81ffc11c212fa327657cb19dd85eb7419e163b5b076bede2bdb5c974c07e4"
-"checksum num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)" = "76dac5ed2a876980778b8b85f75a71b6cbf0db0b1232ee12f826bccb00d09d72"
+"checksum num_cpus 1.13.0 (registry+https://github.com/rust-lang/crates.io-index)" = "05499f3756671c15885fee9034446956fff3f243d6077b91e5767df161f766b3"
 "checksum once_cell 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "532c29a261168a45ce28948f9537ddd7a5dd272cc513b3017b1e82a88f962c37"
 "checksum opaque-debug 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2839e79665f131bdb5782e51f2c6c9599c133c6098982a54c794358bf432529c"
 "checksum openssl-probe 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "77af24da69f9d9341038eba93a073b1fdaaa1b788221b00a69bce9e762cb32de"
@@ -2809,6 +2820,7 @@ dependencies = [
 "checksum textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060"
 "checksum thread-id 3.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "c7fbf4c9d56b320106cd64fd024dadfa0be7cb4706725fc44a7d7ce952d820c1"
 "checksum thread_local 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)" = "c6b53e329000edc2b34dbe8545fd20e55a333362d0a321909685a19bd28c3f1b"
+"checksum threadpool 1.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "e8dae184447c15d5a6916d973c642aec485105a13cd238192a6927ae3e077d66"
 "checksum time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)" = "db8dcfca086c1143c9270ac42a2bbd8a7ee477b78ac8e45b19abfb0cbede4b6f"
 "checksum tiny-bip39 0.6.2 (registry+https://github.com/rust-lang/crates.io-index)" = "c1c5676413eaeb1ea35300a0224416f57abc3bd251657e0fafc12c47ff98c060"
 "checksum tokio 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "0e1bef565a52394086ecac0a6fa3b8ace4cb3a138ee1d96bd2b93283b56824e3"

lib/Cargo.toml

@@ -23,6 +23,9 @@ rand = "0.7.2"
 sodiumoxide = "0.2.5"
 ring = "0.16.9"
 libflate = "0.1"
+subtle = "2"
+threadpool = "1.8.0"
+num_cpus = "1.13.0"
 
 tonic = { version = "0.1.1", features = ["tls", "tls-roots"] }
 bytes = "0.4"

lib/src/grpcconnector.rs

@@ -2,12 +2,15 @@ use log::{error};
 use std::sync::Arc;
 use zcash_primitives::transaction::{TxId};
 
-use crate::grpc_client::{ChainSpec, BlockId, BlockRange, RawTransaction, 
+use crate::grpc_client::{ChainSpec, BlockId, BlockRange, RawTransaction, CompactBlock,
                          TransparentAddressBlockFilter, TxFilter, Empty, LightdInfo, Coinsupply};
 use tonic::transport::{Channel, ClientTlsConfig};
 use tokio_rustls::{rustls::ClientConfig};
 use tonic::{Request};
 
+use threadpool::ThreadPool;
+use std::sync::mpsc::channel;
+
 use crate::PubCertificate;
 use crate::grpc_client::compact_tx_streamer_client::CompactTxStreamerClient;
 
@@ -95,7 +98,7 @@ pub fn get_coinsupply(uri: http::Uri, no_cert: bool) -> Result<Coinsupply, Strin
    // tokio::runtime::current_thread::Runtime::new().unwrap().block_on(runner)
 }
 
-async fn get_block_range<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, c: F) 
+async fn get_block_range<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, pool: ThreadPool, c: F) 
     -> Result<(), Box<dyn std::error::Error>> 
 where F : Fn(&[u8], u64) {
     let mut client = get_client(uri, no_cert).await?;
@@ -105,9 +108,16 @@ where F : Fn(&[u8], u64) {
 
     let request = Request::new(BlockRange{ start: Some(bs), end: Some(be) });
 
-    let mut response = client.get_block_range(request).await?.into_inner();
+     // Channel where the blocks are sent. A None signifies end of all blocks
-    //println!("{:?}", response);
+     let (tx, rx) = channel::<Option<CompactBlock>>();
-    while let Some(block) = response.message().await? {
+
+     // Channel that the processor signals it is done, so the method can return
+     let (ftx, frx) = channel();
+ 
+     // The processor runs on a different thread, so that the network calls don't
+     // block on this
+     pool.execute(move || {
+         while let Some(block) = rx.recv().unwrap() {
              use prost::Message;
              let mut encoded_buf = vec![];
  
@@ -115,10 +125,23 @@ where F : Fn(&[u8], u64) {
              c(&encoded_buf, block.height);
          }
  
+         ftx.send(Ok(())).unwrap();
+     });
+
+    let mut response = client.get_block_range(request).await?.into_inner();
+    //println!("{:?}", response);
+    while let Some(block) = response.message().await? {
+        tx.send(Some(block)).unwrap();
+    }
+    tx.send(None).unwrap();
+
+    // Wait for the processor to exit
+    frx.iter().take(1).collect::<Result<Vec<()>, String>>()?;
+
     Ok(())
 }
 
-pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, c: F) -> Result<(), String>
+pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, pool: ThreadPool, c: F) -> Result<(), String>
     where F : Fn(&[u8], u64) {
     
     let mut rt = match tokio::runtime::Runtime::new() {
@@ -131,7 +154,7 @@ pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_heig
         }
     };
 
-    match rt.block_on(get_block_range(uri, start_height, end_height, no_cert, c)) {
+    match rt.block_on(get_block_range(uri, start_height, end_height, no_cert, pool, c)) {
         Ok(o) => Ok(o),
         Err(e) => {
             let e = format!("Error fetching blocks {:?}", e);
@@ -202,26 +225,26 @@ async fn get_transaction(uri: &http::Uri, txid: TxId, no_cert: bool)
     Ok(response.into_inner())
 }
 
-pub fn fetch_full_tx<F : 'static + std::marker::Send>(uri: &http::Uri, txid: TxId, no_cert: bool, c: F)
+pub fn fetch_full_tx(uri: &http::Uri, txid: TxId, no_cert: bool) -> Result<Vec<u8>, String> {
-        where F : Fn(&[u8]) {
     let mut rt = match tokio::runtime::Runtime::new() {
         Ok(r) => r,
         Err(e) => {
-            error!("Error creating runtime {}", e.to_string());
+            let errstr = format!("Error creating runtime {}", e.to_string());
-            eprintln!("{}", e);
+            error!("{}", errstr);
-            return;
+            eprintln!("{}", errstr);
+            return Err(errstr);
         }
     };
 
     match rt.block_on(get_transaction(uri, txid, no_cert)) {
-        Ok(rawtx) => c(&rawtx.data),
+        Ok(rawtx) => Ok(rawtx.data.to_vec()),
         Err(e) => {
-            error!("Error in get_transaction runtime {}", e.to_string());
+            let errstr = format!("Error in get_transaction runtime {}", e.to_string());
-            eprintln!("{}", e);
+            error!("{}", errstr);
+            eprintln!("{}", errstr);
+            Err(errstr)
         }
     }    
-
-    
 }
 
 // send_transaction GRPC call
@@ -262,22 +285,19 @@ async fn get_latest_block(uri: &http::Uri, no_cert: bool) -> Result<BlockId, Box
     Ok(response.into_inner())
 }
 
-pub fn fetch_latest_block<F : 'static + std::marker::Send>(uri: &http::Uri, no_cert: bool, mut c : F) 
+pub fn fetch_latest_block(uri: &http::Uri, no_cert: bool) -> Result<BlockId, String> {
-    where F : FnMut(BlockId) {
     let mut rt = match tokio::runtime::Runtime::new() {
         Ok(r) => r,
         Err(e) => {
-            error!("Error creating runtime {}", e.to_string());
+            let errstr = format!("Error creating runtime {}", e.to_string());
-            eprintln!("{}", e);
+            eprintln!("{}", errstr);
-            return;
+            return Err(errstr);
         }
     };
 
-    match rt.block_on(get_latest_block(uri, no_cert)) {
+    rt.block_on(get_latest_block(uri, no_cert)).map_err(|e| {
-        Ok(b) => c(b),
+        let errstr = format!("Error getting latest block {}", e.to_string());
-        Err(e) => {
+        eprintln!("{}", errstr);
-            error!("Error getting latest block {}", e.to_string());
+        errstr
-            eprintln!("{}", e);
+    })
-        }
-    };
 }

lib/src/lightclient.rs

@@ -2,17 +2,21 @@ use crate::lightwallet::LightWallet;
 
 use rand::{rngs::OsRng, seq::SliceRandom};
 
-use std::sync::{Arc, RwLock, Mutex};
+use std::sync::{Arc, RwLock, Mutex, mpsc::channel};
-use std::sync::atomic::{AtomicU64, AtomicI32, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicI32, AtomicUsize, Ordering};
 use std::path::{Path, PathBuf};
 use std::fs::File;
 use std::collections::HashMap;
+use std::cmp::{max, min};
 use std::io;
 use std::io::prelude::*;
 use std::io::{BufReader, BufWriter, Error, ErrorKind};
 
 use protobuf::parse_from_bytes;
 
+
+use threadpool::ThreadPool;
+
 use json::{object, array, JsonValue};
 use zcash_primitives::transaction::{TxId, Transaction};
 use zcash_client_backend::{
@@ -30,7 +34,6 @@ use log4rs::append::rolling_file::policy::compound::{
     roll::fixed_window::FixedWindowRoller,
 };
 
-use crate::grpc_client::{BlockId};
 use crate::grpcconnector::{self, *};
 use crate::SaplingParams;
 
@@ -991,14 +994,8 @@ impl LightClient {
         let mut last_scanned_height = self.wallet.read().unwrap().last_scanned_height() as u64;
 
         // This will hold the latest block fetched from the RPC
-        let latest_block_height = Arc::new(AtomicU64::new(0));
-        let lbh = latest_block_height.clone();
-        fetch_latest_block(&self.get_server_uri(), self.config.no_cert_verification, 
-            move |block: BlockId| {
-                lbh.store(block.height, Ordering::SeqCst);
-            });
-        let latest_block = latest_block_height.load(Ordering::SeqCst);
 
+        let latest_block = fetch_latest_block(&self.get_server_uri(), self.config.no_cert_verification)?.height;
        
         if latest_block < last_scanned_height {
             let w = format!("Server's latest block({}) is behind ours({})", latest_block, last_scanned_height);
@@ -1035,6 +1032,9 @@ impl LightClient {
         // belong to us.
         let all_new_txs = Arc::new(RwLock::new(vec![]));
 
+        // Create a new threadpool (upto 8, atleast 2 threads) to scan with
+        let pool = ThreadPool::new(max(2, min(8, num_cpus::get())));
+
         // Fetch CompactBlocks in increments
         let mut pass = 0;
         loop {
@@ -1070,7 +1070,8 @@ impl LightClient {
 
             let last_invalid_height = Arc::new(AtomicI32::new(0));
             let last_invalid_height_inner = last_invalid_height.clone();
-            fetch_blocks(&self.get_server_uri(), start_height, end_height, self.config.no_cert_verification,
+            let tpool = pool.clone();
+            fetch_blocks(&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 {
@@ -1088,7 +1089,7 @@ impl LightClient {
                         Err(_) => {}
                     }
 
-                    match local_light_wallet.read().unwrap().scan_block(encoded_block) {
+                    match local_light_wallet.read().unwrap().scan_block_with_pool(encoded_block, &tpool) {
                         Ok(block_txns) => {
                             // Add to global tx list
                             all_txs.write().unwrap().extend_from_slice(&block_txns.iter().map(|txid| (txid.clone(), height as i32)).collect::<Vec<_>>()[..]);
@@ -1102,6 +1103,16 @@ impl LightClient {
                     local_bytes_downloaded.fetch_add(encoded_block.len(), Ordering::SeqCst);
             })?;
 
+
+            {
+                // println!("Total scan duration: {:?}", self.wallet.read().unwrap().total_scan_duration.read().unwrap().get(0).unwrap().as_millis());
+
+                let t = self.wallet.read().unwrap();
+                let mut d = t.total_scan_duration.write().unwrap();
+                d.clear();
+                d.push(std::time::Duration::new(0, 0));
+            }
+
             // Check if there was any invalid block, which means we might have to do a reorg
             let invalid_height = last_invalid_height.load(Ordering::SeqCst);
             if invalid_height > 0 {
@@ -1136,6 +1147,11 @@ impl LightClient {
                 let addresses = self.wallet.read().unwrap()
                                     .taddresses.read().unwrap().iter().map(|a| a.clone())
                                     .collect::<Vec<String>>();
+
+                 // Create a channel so the fetch_transparent_txids can send the results back
+                 let (ctx, crx) = channel();
+                 let num_addresses = addresses.len();
+
                 for address in addresses {
                     let wallet = self.wallet.clone();
                     let block_times_inner = block_times.clone();
@@ -1148,16 +1164,29 @@ impl LightClient {
                         start_height
                     };
 
-                    fetch_transparent_txids(&self.get_server_uri(), address, transparent_start_height, end_height, self.config.no_cert_verification, 
+                    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 || {
+                        // Fetch the transparent transactions for this address, and send the results 
+                        // via the channel
+                        let r = fetch_transparent_txids(&server_uri, address, transparent_start_height, end_height, no_cert,
                             move |tx_bytes: &[u8], height: u64| {
                                 let tx = Transaction::read(tx_bytes).unwrap();
 
                                 // Scan this Tx for transparent inputs and outputs
                                 let datetime = block_times_inner.read().unwrap().get(&height).map(|v| *v).unwrap_or(0);
                                 wallet.read().unwrap().scan_full_tx(&tx, height as i32, datetime as u64); 
+                        });
+                        ctx.send(r).unwrap();
+                    });
                 }
-                    )?;
+
-                }
+                // Collect all results from the transparent fetches, and make sure everything was OK. 
+                // If it was not, we return an error, which will go back to the retry
+                crx.iter().take(num_addresses).collect::<Result<Vec<()>, String>>()?;
             }           
             
             // Do block height accounting
@@ -1200,24 +1229,44 @@ impl LightClient {
         let mut rng = OsRng;        
         txids_to_fetch.shuffle(&mut rng);
 
+        let num_fetches = txids_to_fetch.len();
+        let (ctx, crx) = channel();
+
         // 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);
 
-            fetch_full_tx(&self.get_server_uri(), txid, self.config.no_cert_verification, move |tx_bytes: &[u8] | {
+                    match fetch_full_tx(&server_uri, txid, no_cert) {
-                let tx = Transaction::read(tx_bytes).unwrap();
+                        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()
+                    };   
             });
         };
 
-        Ok(object!{
+            // Wait for all the fetches to finish.
+            let result = crx.iter().take(num_fetches).collect::<Result<Vec<()>, String>>();
+            match result {
+                Ok(_) => Ok(object!{
                     "result" => "success",
                     "latest_block" => latest_block,
                     "downloaded_bytes" => bytes_downloaded.load(Ordering::SeqCst)
-        })
+                }),
+                Err(e) => Err(format!("Error fetching all txns for memos: {}", e))
+            }   
     }
 
     pub fn do_send(&self, addrs: Vec<(&str, u64, Option<String>)>) -> Result<String, String> {
@@ -1228,16 +1277,18 @@ impl LightClient {
 
         info!("Creating transaction");
 
-        let rawtx = self.wallet.write().unwrap().send_to_address(
+        let result = {
+            let _lock = self.sync_lock.lock().unwrap();
+
+            self.wallet.write().unwrap().send_to_address(
                 u32::from_str_radix(&self.config.consensus_branch_id, 16).unwrap(), 
                 &self.sapling_spend, &self.sapling_output,
-            addrs
+                addrs,
-        );
+                |txbytes| broadcast_raw_tx(&self.get_server_uri(), self.config.no_cert_verification, txbytes)
+            )
+        };
         
-        match rawtx {
+        result.map(|(txid, _)| txid)
-            Ok(txbytes)   => broadcast_raw_tx(&self.get_server_uri(), self.config.no_cert_verification, txbytes),
-            Err(e)        => Err(format!("Error: No Tx to broadcast. Error was: {}", e))
-        }
     }
 }
 

lib/src/lightclient/checkpoints.rs

@@ -76,6 +76,20 @@ fn get_main_checkpoint(height: u64) ->  Option<(u64, &'static str, &'static str)
 
 "012713b68800b2b26a90edddde497f39574fc79fcbbd3abbf5201299e92b24982a00100001900dda253ef5c886f9f5b1d4ec6208e13be23db48f34643995b4710f7c9b154e0001157f4d00c85a744085be70a23bff48fb9fedb5450d26f71dc188b55589ce9b4a000001e695cca82b5f79d2efd87e970a9d59d08f0e38933d5845fc5f65c0ae04904f520001ca0d575f716680bb4c37c600f46db9aec62008fff03b75aab3030f5151f80a540001ce2cfbe86a3b6da4f13471aedc066fdc686b6d89b0a857c3c788d30adebbe86e018b9a1cb81bc08041d80850f2da32920c6f41c99746807e9d3b2c4c67756e2e6e012cadfdc5952c78f13b2f5b67f544f6f4b2fc6a4a87227be96196d3da93aba15701aa689be876025869ea246310586d5c6585fc15b7013520bfe3ab15301adbb72201bbab4f57b3298599502cb7fe6f08271006fd8c6f45ebcdbba4db8a33efe42a370130195f8f73dc864a32004ff13f5fa2fed32723a150a498f593590ccec493e004"
 ),
+
+(270000,  "000000026cc545eed18b508c3368cd20256c012bfa10f5f115b21ad0101c02cb",
+
+"01994f552eccb1c7e9d9eb883627d062792db872b8201a2ec2ce3620fbd7a3c757001101d060165f1156bc1274e7421b1c386cf38e1a1a0022ecd2856af9930fe5b9ce4f00011ca156cf71adc4122768deb7497fef31a45a38daccfc68d7cd04f4b2a4394f6601e4ad07bc4edd97271886a33bba67bcb7d013027b54c1728f4192ceb7b01c02660136690f3b4c30ee836802c2efc70b241896d26007b20316479ecc42716d3d421500000000016644a968a2e9fbba4252a811c46431dbabfa8cc88881a991e6470068a461fd240001ddbdab31cf067f4478e78adb9a03991d1f62ecf171cbf5f19e65671ad84d016a000000000199d67314660501d57c69466623f01bea8f1ba1fd96f9fceb98d545e7632d9841"
+),
+
+(280000,  "000000036b2c0edb762736b4243cdba4d5b576456cc4c6b6a29ed69d27f0c4d9",
+
+"0107f480921bf7e88d0a947cf37937cc2f049e9c02ec9fce955bbbdb1cc1ad4b18001101dea0aa086d1db98a47a9a5eca155bb61a1fee3f0fd04addfc064cf56849b56610145ce19f2de30c816e731e958658013552ac7b2a6ab8c4fc627a2de46772f9e2f019f67c118f1982e4d7f4db835d9e8757b277b55add36d299e5738f93c3f253f2f014d2532531f8b3a2ec1337540befb49d67f97776427fc9457a18a50136f1965470150ca973901b35086d97382fd46214bf539e76347a637d24c1658a919fb00d7710001a10249d1d0356a5742565562bdc9e2d58f6db0322fecd7d9b19d5983dbad0444019d43d46a0f5210cc9cc1d5e1f5e6dcd1a746d64e6b76ac71b0a256124655647201128bcdf48b18f9d4093a7b2b0d952e776671f0c5ba04ea42f9d6bb53a33a7e34019a1b7b861ddaf162f8bfd6021989d3283f983c229b84dbf98f2455a0957e1d09000001e4eb56b101d0d5966f1fea09f426680440aa847debd0df4c24bbcb9e745b7b040000000199d67314660501d57c69466623f01bea8f1ba1fd96f9fceb98d545e7632d9841"
+),
+(290000,  "00000000c9bd5248099f4caca2a5b1da88548cd1824bb22a0efa6c30cf6ccfce",
+
+"014dfd7871fcfe6dd86dd1e61f67a7a0947127eeea52248449974ef0381365c42301d0488b8b1e3ff3e65dead63d9c1352575352a0921e8072f679a513e62bd7344511000001f2922fb7e1c2b071e3281ed70b532d03d3e09cb0a8b4273b001bc3ac4026ce48000001570a88b589167eaa5e8a5b446fdf6bd3aa4936157acb790b81c6ae46f2359f6701f505693c3478a50e3de6ee934592b09e2854071d4ecb27209474ef86adebdd5201c76e39bb424abf87767a048b6b4037a335a92fe72e2ab0aab0955636f3946120019e79f50fb85eb3440a5f1d9ce7fbd28b7ead8fff7c8327ccb60f7615d7abc671000169685f032b7645d1f91a8e5607ce3e051bd3fddb793da404d82beafa40aa701c013925686dae0e4aa89561983dc4ab0b8e94ba75571f41edad029f3c47401a2d4d01e4eb56b101d0d5966f1fea09f426680440aa847debd0df4c24bbcb9e745b7b040000000199d67314660501d57c69466623f01bea8f1ba1fd96f9fceb98d545e7632d9841"
+),
 ];
 
     find_checkpoint(height, checkpoints)

lib/src/lightwallet.rs

@@ -1,11 +1,15 @@
-use std::time::SystemTime;
+use std::time::{SystemTime, Duration};
 use std::io::{self, Read, Write};
 use std::cmp;
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, RwLock};
 use std::io::{Error, ErrorKind};
 
+use threadpool::ThreadPool;
+use std::sync::mpsc::{channel};
+
 use rand::{Rng, rngs::OsRng};
+use subtle::{ConditionallySelectable, ConstantTimeEq, CtOption};
 
 use log::{info, warn, error};
 
@@ -21,20 +25,23 @@ use sha2::{Sha256, Digest};
 
 use zcash_client_backend::{
     encoding::{encode_payment_address, encode_extended_spending_key},
-    proto::compact_formats::CompactBlock, welding_rig::scan_block,
+    proto::compact_formats::{CompactBlock, CompactOutput},
+    wallet::{WalletShieldedOutput, WalletShieldedSpend}
 };
 
 use zcash_primitives::{
+    jubjub::fs::Fs,
     block::BlockHash,
-    merkle_tree::{CommitmentTree},
     serialize::{Vector},
     transaction::{
         builder::{Builder},
         components::{Amount, OutPoint, TxOut}, components::amount::DEFAULT_FEE,
         TxId, Transaction, 
     },
+    sapling::Node,
+    merkle_tree::{CommitmentTree, IncrementalWitness},
     legacy::{Script, TransparentAddress},
-    note_encryption::{Memo, try_sapling_note_decryption, try_sapling_output_recovery},
+    note_encryption::{Memo, try_sapling_note_decryption, try_sapling_output_recovery, try_sapling_compact_note_decryption},
     zip32::{ExtendedFullViewingKey, ExtendedSpendingKey, ChildIndex},
     JUBJUB,
     primitives::{PaymentAddress},
@@ -136,6 +143,8 @@ pub struct LightWallet {
 
     // Non-serialized fields
     config: LightClientConfig,
+
+    pub total_scan_duration: Arc<RwLock<Vec<Duration>>>,
 }
 
 impl LightWallet {
@@ -254,6 +263,7 @@ impl LightWallet {
             mempool_txs: Arc::new(RwLock::new(HashMap::new())),
             config:      config.clone(),
             birthday:    latest_block,
+            total_scan_duration: Arc::new(RwLock::new(vec![Duration::new(0, 0)])),
         };
 
         // If restoring from seed, make sure we are creating 50 addresses for users
@@ -375,6 +385,7 @@ impl LightWallet {
             mempool_txs: Arc::new(RwLock::new(HashMap::new())),
             config:      config.clone(),
             birthday,
+            total_scan_duration: Arc::new(RwLock::new(vec![Duration::new(0, 0)])),
         })
     }
 
@@ -424,12 +435,19 @@ impl LightWallet {
 
         Vector::write(&mut writer, &self.blocks.read().unwrap(), |w, b| b.write(w))?;
                 
-        // The hashmap, write as a set of tuples
+        // The hashmap, write as a set of tuples. Store them sorted so that wallets are
-        Vector::write(&mut writer, &self.txs.read().unwrap().iter().collect::<Vec<(&TxId, &WalletTx)>>(),
+        // deterministically saved
+        {
+            let txlist = self.txs.read().unwrap();
+            let mut txns = txlist.iter().collect::<Vec<(&TxId, &WalletTx)>>();
+            txns.sort_by(|a, b| a.0.partial_cmp(b.0).unwrap());
+
+            Vector::write(&mut writer, &txns,
                             |w, (k, v)| {
                                 w.write_all(&k.0)?;
                                 v.write(w)
                             })?;
+        }
         utils::write_string(&mut writer, &self.config.chain_name)?;
 
         // While writing the birthday, get it from the fn so we recalculate it properly
@@ -1272,7 +1290,7 @@ impl LightWallet {
             // Trim all witnesses for the invalidated blocks
             for tx in txs.values_mut() {
                 for nd in tx.notes.iter_mut() {
-                    nd.witnesses.split_off(nd.witnesses.len().saturating_sub(num_invalidated));
+                    let _discard = nd.witnesses.split_off(nd.witnesses.len().saturating_sub(num_invalidated));
                 }
             }
         }
@@ -1280,8 +1298,234 @@ impl LightWallet {
         num_invalidated as u64
     }
 
-    // Scan a block. Will return an error with the block height that failed to scan
+       /// Scans a [`CompactOutput`] with a set of [`ExtendedFullViewingKey`]s.
+    ///
+    /// Returns a [`WalletShieldedOutput`] and corresponding [`IncrementalWitness`] if this
+    /// output belongs to any of the given [`ExtendedFullViewingKey`]s.
+    ///
+    /// The given [`CommitmentTree`] and existing [`IncrementalWitness`]es are incremented
+    /// with this output's commitment.
+    fn scan_output_internal(
+        &self,
+        (index, output): (usize, CompactOutput),
+        ivks: &[Fs],
+        tree: &mut CommitmentTree<Node>,
+        existing_witnesses: &mut [&mut IncrementalWitness<Node>],
+        block_witnesses: &mut [&mut IncrementalWitness<Node>],
+        new_witnesses: &mut [&mut IncrementalWitness<Node>],
+        pool: &ThreadPool
+    ) -> Option<WalletShieldedOutput> {
+        let cmu = output.cmu().ok()?;
+        let epk = output.epk().ok()?;
+        let ct = output.ciphertext;
+
+        let (tx, rx) = channel();
+        ivks.iter().enumerate().for_each(|(account, ivk)| {
+            // Clone all values for passing to the closure
+            let ivk = ivk.clone();
+            let epk = epk.clone();
+            let ct = ct.clone();
+            let tx = tx.clone();
+
+            pool.execute(move || {
+                let m = try_sapling_compact_note_decryption(&ivk, &epk, &cmu, &ct);
+                let r = match m {
+                    Some((note, to)) => {
+                        tx.send(Some(Some((note, to, account))))
+                    },
+                    None => {
+                        tx.send(Some(None))
+                    }
+                };
+
+                match r {
+                    Ok(_) => {},
+                    Err(e) => println!("Send error {:?}", e)
+                }
+            });
+        });
+
+        // Increment tree and witnesses
+        let node = Node::new(cmu.into());
+        for witness in existing_witnesses {
+            witness.append(node).unwrap();
+        }
+        for witness in block_witnesses {
+            witness.append(node).unwrap();
+        }
+        for witness in new_witnesses {
+            witness.append(node).unwrap();
+        }
+        tree.append(node).unwrap();
+
+        // Collect all the RXs and fine if there was a valid result somewhere
+        let mut wsos =  vec![];
+        for _i in 0..ivks.len() {
+            let n = rx.recv().unwrap();
+            let epk = epk.clone();
+
+            let wso = match n {
+                None => panic!("Got a none!"),
+                Some(None) => None,
+                Some(Some((note, to, account))) => {
+                    // A note is marked as "change" if the account that received it
+                    // also spent notes in the same transaction. This will catch,
+                    // for instance:
+                    // - Change created by spending fractions of notes.
+                    // - Notes created by consolidation transactions.
+                    // - Notes sent from one account to itself.
+                    //let is_change = spent_from_accounts.contains(&account);
+
+                    Some(WalletShieldedOutput {
+                        index, cmu, epk, account, note, to, is_change: false,
+                        witness: IncrementalWitness::from_tree(tree),
+                    })
+                }
+            };
+            wsos.push(wso);
+        }
+
+        match wsos.into_iter().find(|wso| wso.is_some()) {
+            Some(Some(wso)) => Some(wso),
+            _ => None
+        }
+    }
+
+    /// Scans a [`CompactBlock`] with a set of [`ExtendedFullViewingKey`]s.
+    ///
+    /// Returns a vector of [`WalletTx`]s belonging to any of the given
+    /// [`ExtendedFullViewingKey`]s, and the corresponding new [`IncrementalWitness`]es.
+    ///
+    /// The given [`CommitmentTree`] and existing [`IncrementalWitness`]es are
+    /// incremented appropriately.
+    pub fn scan_block_internal(
+        &self,
+        block: CompactBlock,
+        extfvks: &[ExtendedFullViewingKey],
+        nullifiers: Vec<(Vec<u8>, usize)>,
+        tree: &mut CommitmentTree<Node>,
+        existing_witnesses: &mut [&mut IncrementalWitness<Node>],
+        pool: &ThreadPool
+    ) -> Vec<zcash_client_backend::wallet::WalletTx> {
+        let mut wtxs: Vec<zcash_client_backend::wallet::WalletTx> = vec![];
+        let ivks = extfvks.iter().map(|extfvk| extfvk.fvk.vk.ivk()).collect::<Vec<_>>();
+
+        for tx in block.vtx.into_iter() {
+            let num_spends = tx.spends.len();
+            let num_outputs = tx.outputs.len();
+
+            let (ctx, crx) = channel();
+            {
+                let nullifiers = nullifiers.clone();
+                let tx = tx.clone();
+                pool.execute(move || {
+                    // Check for spent notes
+                    // The only step that is not constant-time is the filter() at the end.
+                    let shielded_spends: Vec<_> = tx
+                        .spends
+                        .into_iter()
+                        .enumerate()
+                        .map(|(index, spend)| {
+                            // Find the first tracked nullifier that matches this spend, and produce
+                            // a WalletShieldedSpend if there is a match, in constant time.
+                            nullifiers
+                                .iter()
+                                .map(|(nf, account)| CtOption::new(*account as u64, nf.ct_eq(&spend.nf[..])))
+                                .fold(CtOption::new(0, 0.into()), |first, next| {
+                                    CtOption::conditional_select(&next, &first, first.is_some())
+                                })
+                                .map(|account| WalletShieldedSpend {
+                                    index,
+                                    nf: spend.nf,
+                                    account: account as usize,
+                                })
+                        })
+                        .filter(|spend| spend.is_some().into())
+                        .map(|spend| spend.unwrap())
+                        .collect();
+
+                    // Collect the set of accounts that were spent from in this transaction
+                    let spent_from_accounts: HashSet<_> =
+                        shielded_spends.iter().map(|spend| spend.account).collect();
+
+                    ctx.send((shielded_spends, spent_from_accounts)).unwrap();
+
+                    drop(ctx);
+                });
+            }
+
+
+            // Check for incoming notes while incrementing tree and witnesses
+            let mut shielded_outputs: Vec<WalletShieldedOutput> = vec![];
+            {
+                // Grab mutable references to new witnesses from previous transactions
+                // in this block so that we can update them. Scoped so we don't hold
+                // mutable references to wtxs for too long.
+                let mut block_witnesses: Vec<_> = wtxs
+                    .iter_mut()
+                    .map(|tx| {
+                        tx.shielded_outputs
+                            .iter_mut()
+                            .map(|output| &mut output.witness)
+                    })
+                    .flatten()
+                    .collect();
+
+                for to_scan in tx.outputs.into_iter().enumerate() {
+                    // Grab mutable references to new witnesses from previous outputs
+                    // in this transaction so that we can update them. Scoped so we
+                    // don't hold mutable references to shielded_outputs for too long.
+                    let mut new_witnesses: Vec<_> = shielded_outputs
+                        .iter_mut()
+                        .map(|output| &mut output.witness)
+                        .collect();
+
+                    if let Some(output) = self.scan_output_internal(
+                        to_scan,
+                        &ivks,
+                        tree,
+                        existing_witnesses,
+                        &mut block_witnesses,
+                        &mut new_witnesses,
+                        pool
+                    ) {
+                        shielded_outputs.push(output);
+                    }
+                }
+            }
+
+            let (shielded_spends, spent_from_accounts) = crx.recv().unwrap();
+
+            // Identify change outputs
+            shielded_outputs.iter_mut().for_each(|output| {
+                if spent_from_accounts.contains(&output.account) {
+                    output.is_change = true;
+                }
+            });
+
+            // Update wallet tx
+            if !(shielded_spends.is_empty() && shielded_outputs.is_empty()) {
+                let mut txid = TxId([0u8; 32]);
+                txid.0.copy_from_slice(&tx.hash);
+                wtxs.push(zcash_client_backend::wallet::WalletTx {
+                    txid,
+                    index: tx.index as usize,
+                    num_spends,
+                    num_outputs,
+                    shielded_spends,
+                    shielded_outputs,
+                });
+            }
+        }
+
+        wtxs
+    }
     pub fn scan_block(&self, block_bytes: &[u8]) -> Result<Vec<TxId>, i32> {
+        self.scan_block_with_pool(&block_bytes, &ThreadPool::new(1))
+    }
+
+    // Scan a block. Will return an error with the block height that failed to scan
+    pub fn scan_block_with_pool(&self, block_bytes: &[u8], pool: &ThreadPool) -> Result<Vec<TxId>, i32> {
         let block: CompactBlock = match parse_from_bytes(block_bytes) {
             Ok(block) => block,
             Err(e) => {
@@ -1372,7 +1616,7 @@ impl LightWallet {
             }
 
             new_txs = {
-                let nf_refs: Vec<_> = nfs.iter().map(|(nf, acc, _)| (&nf[..], *acc)).collect();
+                let nf_refs = nfs.iter().map(|(nf, account, _)| (nf.to_vec(), *account)).collect::<Vec<_>>();
 
                 // Create a single mutable slice of all the newly-added witnesses.
                 let mut witness_refs: Vec<_> = txs
@@ -1381,12 +1625,13 @@ impl LightWallet {
                     .flatten()
                     .collect();
 
-                scan_block(
+                self.scan_block_internal(
                     block.clone(),
                     &self.extfvks.read().unwrap(),
-                    &nf_refs[..],
+                    nf_refs,
                     &mut block_data.tree,
                     &mut witness_refs[..],
+                    pool,
                 )
             };
         }
@@ -1492,13 +1737,16 @@ impl LightWallet {
         Ok(all_txs)
     }
 
-    pub fn send_to_address(
+    pub fn send_to_address<F> (
         &self,
         consensus_branch_id: u32,
         spend_params: &[u8],
         output_params: &[u8],
-        tos: Vec<(&str, u64, Option<String>)>
+        tos: Vec<(&str, u64, Option<String>)>,
-    ) -> Result<Box<[u8]>, String> {
+        broadcast_fn: F
+    ) -> Result<(String, Vec<u8>), String> 
+        where F: Fn(Box<[u8]>) -> Result<String, String>
+    {
         if !self.unlocked {
             return Err("Cannot spend while wallet is locked".to_string());
         }
@@ -1557,12 +1805,19 @@ impl LightWallet {
         // Select notes to cover the target value
         println!("{}: Selecting notes", now() - start_time);
         let target_value = Amount::from_u64(total_value).unwrap() + DEFAULT_FEE ;
-        let notes: Vec<_> = self.txs.read().unwrap().iter()
+          // Select the candidate notes that are eligible to be spent
+        let mut candidate_notes: Vec<_> = self.txs.read().unwrap().iter()
             .map(|(txid, tx)| tx.notes.iter().map(move |note| (*txid, note)))
             .flatten()
             .filter_map(|(txid, note)|
                 SpendableNote::from(txid, note, anchor_offset, &self.extsks.read().unwrap()[note.account])
-            )
+            ).collect();
+
+            // Sort by highest value-notes first.
+            candidate_notes.sort_by(|a, b| b.note.value.cmp(&a.note.value));
+    
+            // Select the minimum number of notes required to satisfy the target value
+            let notes: Vec<_> = candidate_notes.iter()
             .scan(0, |running_total, spendable| {
                 let value = spendable.note.value;
                 let ret = if *running_total < u64::from(target_value) {
@@ -1706,6 +1961,12 @@ impl LightWallet {
         println!("{}: Transaction created", now() - start_time);
         println!("Transaction ID: {}", tx.txid());
 
+                // Create the TX bytes
+                let mut raw_tx = vec![];
+                tx.write(&mut raw_tx).unwrap();
+        
+                let txid = broadcast_fn(raw_tx.clone().into_boxed_slice())?;
+
         // Mark notes as spent.
         {
             // Mark sapling notes as unconfirmed spent
@@ -1765,10 +2026,7 @@ impl LightWallet {
             }
         }
 
-        // Return the encoded transaction, so the caller can send it.
+        Ok((txid, raw_tx))
-        let mut raw_tx = vec![];
-        tx.write(&mut raw_tx).unwrap();
-        Ok(raw_tx.into_boxed_slice())
     }
 
     // After some blocks have been mined, we need to remove the Txns from the mempool_tx structure

lib/src/lightwallet/tests.rs

@@ -713,8 +713,8 @@ fn test_z_spend_to_z() {
     }
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -828,8 +828,8 @@ fn test_self_txns_ttoz_withmemo() {
     let (ss, so) =get_sapling_params().unwrap();
 
     // Create a tx and send to address. This should consume both the UTXO and the note
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&zaddr, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -922,8 +922,8 @@ fn test_self_txns_ztoz() {
     let (ss, so) =get_sapling_params().unwrap();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&zaddr2, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&zaddr2, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -942,8 +942,8 @@ fn test_self_txns_ztoz() {
     }
 
     // Another self tx, this time without a memo
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-        vec![(&zaddr2, AMOUNT_SENT, None)]).unwrap();
+        vec![(&zaddr2, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
 
@@ -978,8 +978,8 @@ fn test_multi_z() {
     let (ss, so) =get_sapling_params().unwrap();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&zaddr2, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&zaddr2, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -1033,8 +1033,8 @@ fn test_multi_z() {
     let amount_all:u64 = (AMOUNT1 - AMOUNT_SENT - fee) + (AMOUNT_SENT) - fee;
     let taddr = wallet.address_from_sk(&SecretKey::from_slice(&[1u8; 32]).unwrap());
 
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                                        vec![(&taddr, amount_all, None)]).unwrap();
+        vec![(&taddr, amount_all, None)], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_ext_txid = sent_tx.txid();
 
@@ -1075,8 +1075,8 @@ fn test_z_spend_to_taddr() {
     const AMOUNT_SENT: u64 = 30;
     let fee: u64 = DEFAULT_FEE.try_into().unwrap();
 
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                                        vec![(&taddr, AMOUNT_SENT, None)]).unwrap();
+        vec![(&taddr, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
 
@@ -1123,8 +1123,8 @@ fn test_z_spend_to_taddr() {
     }
 
     // Create a new Tx, but this time with a memo.
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-        vec![(&taddr, AMOUNT_SENT, Some("T address memo".to_string()))]).unwrap();
+        vec![(&taddr, AMOUNT_SENT, Some("T address memo".to_string()))], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid2 = sent_tx.txid();
 
@@ -1198,8 +1198,8 @@ fn test_t_spend_to_z() {
     let (ss, so) =get_sapling_params().unwrap();
 
     // Create a tx and send to address. This should consume both the UTXO and the note
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -1272,8 +1272,8 @@ fn test_z_incoming_memo() {
     let (ss, so) = get_sapling_params().unwrap();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&my_address, AMOUNT1 - fee, Some(memo.clone()))]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
 
@@ -1315,8 +1315,8 @@ fn test_add_new_zt_hd_after_incoming() {
     assert_eq!(wallet.zaddress.read().unwrap().len(), 6);   // Starts with 1+5 addresses
 
     // Create a tx and send to the last address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&my_address, AMOUNT1 - fee, None)]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
 
     // Add it to a block
@@ -1358,8 +1358,8 @@ fn test_z_to_t_withinwallet() {
     let (ss, so) = get_sapling_params().unwrap();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&taddr, AMOUNT_SENT, None)]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
 
@@ -1418,8 +1418,8 @@ fn test_multi_t() {
     let (ss, so) = get_sapling_params().unwrap();
 
     // Create a Tx and send to the second t address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&taddr2, AMOUNT_SENT1, None)]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid1 = sent_tx.txid();
 
@@ -1462,8 +1462,8 @@ fn test_multi_t() {
     let taddr3 = wallet.add_taddr();
 
     // Create a Tx and send to the second t address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&taddr3, AMOUNT_SENT2, None)]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid2 = sent_tx.txid();
 
@@ -1499,8 +1499,8 @@ fn test_multi_t() {
     let outgoing_memo = "Outgoing Memo".to_string();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&ext_address, AMOUNT_SENT_EXT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid3 = sent_tx.txid();
@@ -1557,7 +1557,7 @@ fn test_multi_spends() {
                     (taddr2.as_str(), TAMOUNT2, None),
                     (taddr3.as_str(), TAMOUNT3, None) ];
     
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so, tos).unwrap();
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so, tos, |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
 
@@ -1629,7 +1629,7 @@ fn test_multi_spends() {
 
     let tos = vec![ (ext_address.as_str(), EXT_ZADDR_AMOUNT, Some(ext_memo.clone())),
                     (ext_taddr.as_str(), ext_taddr_amount, None)];
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so, tos).unwrap();
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so, tos, |_| Ok(' '.to_string())).unwrap();
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid2 = sent_tx.txid();
 
@@ -1682,16 +1682,16 @@ fn test_bad_send() {
 
     // Bad address
     let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
-                                        vec![(&"badaddress", 10, None)]);
+                                        vec![(&"badaddress", 10, None)], |_| Ok(' '.to_string()));
     assert!(raw_tx.err().unwrap().contains("Invalid recipient address"));
 
     // Insufficient funds
     let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
-                                        vec![(&ext_taddr, AMOUNT1 + 10, None)]);
+                                         vec![(&ext_taddr, AMOUNT1 + 10, None)], |_| Ok(' '.to_string()));
     assert!(raw_tx.err().unwrap().contains("Insufficient verified funds"));
 
     // No addresses
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so, vec![]);
+    let raw_tx = wallet.send_to_address(branch_id, &ss, &so, vec![], |_| Ok(' '.to_string()));
     assert!(raw_tx.err().unwrap().contains("at least one"));
 
 }
@@ -1712,7 +1712,7 @@ fn test_duplicate_outputs() {
     let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
         vec![(&ext_taddr, 100, Some("First memo".to_string())),
              (&ext_taddr, 0, Some("Second memo".to_string())),
-             (&ext_taddr, 0, Some("Third memo".to_string()))]);
+             (&ext_taddr, 0, Some("Third memo".to_string()))], |_| Ok(' '.to_string()));
     assert!(raw_tx.is_ok());
 }
 
@@ -1725,7 +1725,7 @@ fn test_bad_params() {
     let branch_id = u32::from_str_radix("76b809bb", 16).unwrap();
     // Bad params
     let _ = wallet.send_to_address(branch_id, &[], &[],
-                            vec![(&ext_taddr, 10, None)]);
+                                vec![(&ext_taddr, 10, None)], |_| Ok(' '.to_string()));
 }
 
 /// Test helper to add blocks
@@ -1761,8 +1761,8 @@ fn test_z_mempool_expiry() {
     let (ss, so) = get_sapling_params().unwrap();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+                         vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -1878,8 +1878,8 @@ fn test_rollback() {
     // Create a tx and send to address
     const AMOUNT_SENT: u64 = 30000;
     let fee: u64 = DEFAULT_FEE.try_into().unwrap();
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&taddr, AMOUNT_SENT, None)]).unwrap();
+                                vec![(&taddr, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).unwrap();
 
     let sent_tx = Transaction::read(&raw_tx[..]).unwrap();
     let sent_txid = sent_tx.txid();
@@ -2142,8 +2142,8 @@ fn test_encrypted_zreceive() {
     let (ss, so) = get_sapling_params().unwrap();
 
     // Create a tx and send to address
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))]).unwrap();
+        vec![(&ext_address, AMOUNT_SENT, Some(outgoing_memo.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     // Now that we have the transaction, we'll encrypt the wallet
     wallet.encrypt(password.clone()).unwrap();
@@ -2186,7 +2186,7 @@ fn test_encrypted_zreceive() {
 
     // Trying to spend from a locked wallet is an error
     assert!(wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&ext_address, AMOUNT_SENT, None)]).is_err());
+                                vec![(&ext_address, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).is_err());
 
     // unlock the wallet so we can spend to the second z address
     wallet.unlock(password.clone()).unwrap();
@@ -2196,8 +2196,8 @@ fn test_encrypted_zreceive() {
     const ZAMOUNT2:u64 = 30;
     let outgoing_memo2 = "Outgoing Memo2".to_string();
 
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&zaddr2, ZAMOUNT2, Some(outgoing_memo2.clone()))]).unwrap();
+        vec![(&zaddr2, ZAMOUNT2, Some(outgoing_memo2.clone()))], |_| Ok(' '.to_string())).unwrap();
 
     // Now lock the wallet again
     wallet.lock().unwrap();
@@ -2251,8 +2251,8 @@ fn test_encrypted_treceive() {
     const AMOUNT_SENT: u64 = 30;
     let fee: u64 = DEFAULT_FEE.try_into().unwrap();
 
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                                        vec![(&taddr, AMOUNT_SENT, None)]).unwrap();
+        vec![(&taddr, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).unwrap();
 
     // Now that we have the transaction, we'll encrypt the wallet
     wallet.encrypt(password.clone()).unwrap();
@@ -2288,7 +2288,7 @@ fn test_encrypted_treceive() {
 
     // Trying to spend from a locked wallet is an error
     assert!(wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&taddr, AMOUNT_SENT, None)]).is_err());
+                                    vec![(&taddr, AMOUNT_SENT, None)], |_| Ok(' '.to_string())).is_err());
 
     // unlock the wallet so we can spend to the second z address
     wallet.unlock(password.clone()).unwrap();
@@ -2297,8 +2297,8 @@ fn test_encrypted_treceive() {
     let taddr2 = wallet.add_taddr();
     const TAMOUNT2:u64 = 50;
 
-    let raw_tx = wallet.send_to_address(branch_id, &ss, &so,
+    let (_, raw_tx) = wallet.send_to_address(branch_id, &ss, &so,
-                            vec![(&taddr2, TAMOUNT2, None)]).unwrap();
+        vec![(&taddr2, TAMOUNT2, None)], |_| Ok(' '.to_string())).unwrap();
 
     // Now lock the wallet again
     wallet.lock().unwrap();